[net] Add SendPutReq to RCurlConnection for S3 uploads

net/curl/inc/ROOT/RCurlConnection.hxx

@@ -75,6 +75,7 @@ private:
 
    void SetupErrorBuffer();
    void SetOptions();
+   void ResetHandle();
    RResult<void> SetUrl(const std::string &url);
    void Perform(RStatus &status);
 
@@ -116,6 +117,8 @@ public:
    /// a valid batching of requests into multiple multi-range requests takes place automatically.
    /// The fNBytesRecv member of the ranges is only well-defined on success.
    RStatus SendRangesReq(std::size_t N, RUserRange *ranges);
+   /// Uploads data to the URL using an HTTP PUT request.
+   RStatus SendPutReq(const unsigned char *data, std::size_t length);
 
    const std::string &GetEscapedUrl() const { return fEscapedUrl; }
 

net/curl/src/RCurlConnection.cxx

@@ -552,6 +552,61 @@ void ReverseDisplacements(std::vector<std::size_t> &displacements, ROOT::Interna
    }
 }
 
+/// State for the PUT upload read callback: tracks progress through the upload buffer.
+struct RUploadState {
+   const unsigned char *fData = nullptr;
+   std::size_t fLength = 0;
+   std::size_t fOffset = 0;
+};
+
+/// CURLOPT_READFUNCTION callback for PUT uploads.  Copies up to `size * nmemb` bytes from the
+/// upload buffer into `buffer` and advances the offset.  Returns the number of bytes copied,
+/// i.e. min(requested, remaining), or 0 at end-of-data to signal that the upload is complete.
+std::size_t CallbackPutRead(char *buffer, std::size_t size, std::size_t nmemb, void *userdata)
+{
+   auto *state = static_cast<RUploadState *>(userdata);
+   R__ASSERT(state->fOffset <= state->fLength);
+
+   std::size_t remaining = state->fLength - state->fOffset;
+   if (remaining == 0)
+      return 0;
+
+   std::size_t requested = size * nmemb;
+   // CURL_READFUNC_ABORT (0x10000000) collides with a valid byte count at 256 MiB;
+   // assert that curl never asks for that much in a single callback invocation.
+   R__ASSERT(requested < CURL_READFUNC_ABORT);
+   std::size_t nbytes = std::min(requested, remaining);
+   memcpy(buffer, state->fData + state->fOffset, nbytes);
+   state->fOffset += nbytes;
+   return nbytes;
+}
+
+/// CURLOPT_SEEKFUNCTION callback for PUT uploads.  Required because CURLOPT_FOLLOWLOCATION
+/// is enabled: on a redirect curl needs to rewind the upload data before resending.
+/// Returns CURL_SEEKFUNC_OK (0) on success, CURL_SEEKFUNC_FAIL (1) for invalid offsets
+/// which aborts the transfer, or CURL_SEEKFUNC_CANTSEEK (2) for unsupported seek origins
+/// which lets curl try to work around it.
+int CallbackPutSeek(void *userdata, curl_off_t offset, int origin)
+{
+   auto *state = static_cast<RUploadState *>(userdata);
+   // curl documents that it will only use SEEK_SET; guard against anything else defensively.
+   if (origin != SEEK_SET)
+      return CURL_SEEKFUNC_CANTSEEK;
+   if (offset < 0 || static_cast<std::size_t>(offset) > state->fLength)
+      return CURL_SEEKFUNC_FAIL;
+   state->fOffset = static_cast<std::size_t>(offset);
+   return CURL_SEEKFUNC_OK;
+}
+
+/// Wrapper around curl_easy_setopt that asserts on failure.  Most option-setting calls in this
+/// file use valid options and values by construction, so failure indicates a programming error.
+template <typename T>
+void SetCurlOption(void *handle, CURLoption option, T value)
+{
+   auto rc = curl_easy_setopt(handle, option, value);
+   R__ASSERT(rc == CURLE_OK);
+}
+
 std::string GetCurlErrorString(CURLcode code)
 {
    return std::string(curl_easy_strerror(code)) + " (" + std::to_string(code) + ")";
@@ -631,33 +686,43 @@ void ROOT::Internal::RCurlConnection::SetupErrorBuffer()
 {
    if (!fErrorBuffer)
       fErrorBuffer = std::make_unique<char[]>(CURL_ERROR_SIZE);
-   auto rc = curl_easy_setopt(fHandle, CURLOPT_ERRORBUFFER, fErrorBuffer.get());
-   R__ASSERT(rc == CURLE_OK);
+   SetCurlOption(fHandle, CURLOPT_ERRORBUFFER, fErrorBuffer.get());
 }
 
 void ROOT::Internal::RCurlConnection::SetOptions()
 {
-   int rc;
-
    if (gDebug) {
-      rc = curl_easy_setopt(fHandle, CURLOPT_VERBOSE, 1);
-      R__ASSERT(rc == CURLE_OK);
-      rc = curl_easy_setopt(fHandle, CURLOPT_DEBUGFUNCTION, CallbackDebug);
-      R__ASSERT(rc == CURLE_OK);
+      SetCurlOption(fHandle, CURLOPT_VERBOSE, 1);
+      SetCurlOption(fHandle, CURLOPT_DEBUGFUNCTION, CallbackDebug);
    } else {
-      rc = curl_easy_setopt(fHandle, CURLOPT_VERBOSE, 0);
-      R__ASSERT(rc == CURLE_OK);
+      SetCurlOption(fHandle, CURLOPT_VERBOSE, 0);
    }
 
    static const std::string kUserAgent = GetUserAgentString();
-   rc = curl_easy_setopt(fHandle, CURLOPT_USERAGENT, kUserAgent.c_str());
-   R__ASSERT(rc == CURLE_OK);
+   SetCurlOption(fHandle, CURLOPT_USERAGENT, kUserAgent.c_str());
+   SetCurlOption(fHandle, CURLOPT_FOLLOWLOCATION, 1);
+}
 
-   rc = curl_easy_setopt(fHandle, CURLOPT_FOLLOWLOCATION, 1);
-   R__ASSERT(rc == CURLE_OK);
+/// Reset method-specific sticky curl options so that the easy handle is in a clean state
+/// before configuring it for the next request (HEAD, GET, or PUT).
+void ROOT::Internal::RCurlConnection::ResetHandle()
+{
+   SetCurlOption(fHandle, CURLOPT_NOBODY, 0L);
+   SetCurlOption(fHandle, CURLOPT_HTTPGET, 0L);
+   SetCurlOption(fHandle, CURLOPT_UPLOAD, 0L);
+   SetCurlOption(fHandle, CURLOPT_RANGE, static_cast<const char *>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_WRITEFUNCTION, static_cast<curl_write_callback>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_WRITEDATA, static_cast<void *>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_READFUNCTION, static_cast<curl_read_callback>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_READDATA, static_cast<void *>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_SEEKFUNCTION, static_cast<curl_seek_callback>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_SEEKDATA, static_cast<void *>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_INFILESIZE_LARGE, static_cast<curl_off_t>(-1));
 
-   rc = curl_easy_setopt(fHandle, CURLOPT_WRITEFUNCTION, CallbackData);
-   R__ASSERT(rc == CURLE_OK);
+#ifndef HAS_CURL_EASY_HEADER
+   SetCurlOption(fHandle, CURLOPT_HEADERFUNCTION, static_cast<curl_write_callback>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_HEADERDATA, static_cast<void *>(nullptr));
+#endif
 }
 
 ROOT::RResult<void> ROOT::Internal::RCurlConnection::SetUrl(const std::string &url)
@@ -731,23 +796,14 @@ ROOT::Internal::RCurlConnection::RStatus ROOT::Internal::RCurlConnection::SendHe
 {
    remoteSize = kUnknownSize;
 
-   auto rc = curl_easy_setopt(fHandle, CURLOPT_NOBODY, 1);
-   R__ASSERT(rc == CURLE_OK);
-   rc = curl_easy_setopt(fHandle, CURLOPT_RANGE, NULL); // may have been set by a previous SendRangesReq() on the object
-   R__ASSERT(rc == CURLE_OK);
-
-#ifndef HAS_CURL_EASY_HEADER
-   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERFUNCTION, NULL);
-   R__ASSERT(rc == CURLE_OK);
-   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERDATA, NULL);
-   R__ASSERT(rc == CURLE_OK);
-#endif
+   ResetHandle();
+   SetCurlOption(fHandle, CURLOPT_NOBODY, 1);
 
    RStatus status;
    Perform(status);
    if (status) {
       curl_off_t length = -1;
-      rc = curl_easy_getinfo(fHandle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &length);
+      auto rc = curl_easy_getinfo(fHandle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &length);
       if (rc == CURLE_OK && length >= 0)
          remoteSize = length;
    }
@@ -778,18 +834,16 @@ ROOT::Internal::RCurlConnection::SendRangesReq(std::size_t N, RUserRange *ranges
       return RStatus(RStatus::kSuccess);
    }
 
-   auto rc = curl_easy_setopt(fHandle, CURLOPT_HTTPGET, 1);
-   R__ASSERT(rc == CURLE_OK);
+   ResetHandle();
+   SetCurlOption(fHandle, CURLOPT_HTTPGET, 1);
 
    RTransferState transfer(ranges, order, fHandle);
-   rc = curl_easy_setopt(fHandle, CURLOPT_WRITEDATA, &transfer);
-   R__ASSERT(rc == CURLE_OK);
+   SetCurlOption(fHandle, CURLOPT_WRITEFUNCTION, CallbackData);
+   SetCurlOption(fHandle, CURLOPT_WRITEDATA, &transfer);
 
 #ifndef HAS_CURL_EASY_HEADER
-   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERFUNCTION, CallbackHeader);
-   R__ASSERT(rc == CURLE_OK);
-   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERDATA, &transfer);
-   R__ASSERT(rc == CURLE_OK);
+   SetCurlOption(fHandle, CURLOPT_HEADERFUNCTION, CallbackHeader);
+   SetCurlOption(fHandle, CURLOPT_HEADERDATA, &transfer);
 #endif
 
    RStatus status;
@@ -811,8 +865,7 @@ ROOT::Internal::RCurlConnection::SendRangesReq(std::size_t N, RUserRange *ranges
          for (std::size_t i = 1; i < nRanges; ++i) {
             rangeHeader += "," + requestRanges[b + i].ToString();
          }
-         rc = curl_easy_setopt(fHandle, CURLOPT_RANGE, rangeHeader.c_str());
-         R__ASSERT(rc == CURLE_OK);
+         SetCurlOption(fHandle, CURLOPT_RANGE, rangeHeader.c_str());
 
          if (b > 0) {
             const std::uint64_t lastByteRequested = requestRanges[b - 1].fLastByte;
@@ -849,6 +902,26 @@ ROOT::Internal::RCurlConnection::SendRangesReq(std::size_t N, RUserRange *ranges
    return status;
 }
 
+ROOT::Internal::RCurlConnection::RStatus
+ROOT::Internal::RCurlConnection::SendPutReq(const unsigned char *data, std::size_t length)
+{
+   ResetHandle();
+
+   SetCurlOption(fHandle, CURLOPT_UPLOAD, 1L);
+   SetCurlOption(fHandle, CURLOPT_INFILESIZE_LARGE, static_cast<curl_off_t>(length));
+
+   RUploadState uploadState{data, length, 0};
+   SetCurlOption(fHandle, CURLOPT_READFUNCTION, CallbackPutRead);
+   SetCurlOption(fHandle, CURLOPT_READDATA, &uploadState);
+   SetCurlOption(fHandle, CURLOPT_SEEKFUNCTION, CallbackPutSeek);
+   SetCurlOption(fHandle, CURLOPT_SEEKDATA, &uploadState);
+
+   RStatus status;
+   Perform(status);
+
+   return status;
+}
+
 void ROOT::Internal::RCurlConnection::SetCredentials(const RS3Credentials &credentials)
 {
    ClearCredentials();
@@ -876,13 +949,10 @@ void ROOT::Internal::RCurlConnection::ClearCredentials()
    if (!fCredentials)
       return;
 
-   CURLcode rc;
    switch (fCredentials->fType) {
    case EHTTPCredentialsType::kS3:
-      rc = curl_easy_setopt(fHandle, CURLOPT_AWS_SIGV4, NULL);
-      R__ASSERT(rc == CURLE_OK);
-      rc = curl_easy_setopt(fHandle, CURLOPT_USERPWD, NULL);
-      R__ASSERT(rc == CURLE_OK);
+      SetCurlOption(fHandle, CURLOPT_AWS_SIGV4, static_cast<const char *>(nullptr));
+      SetCurlOption(fHandle, CURLOPT_USERPWD, static_cast<const char *>(nullptr));
       break;
    default: R__ASSERT(false && "internal error: unknown credentials type");
    }