client.cc

// MinIO C++ Library for Amazon S3 Compatible Cloud Storage
// Copyright 2022-2024 MinIO, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

#include "miniocpp/client.h"

#ifdef _MSC_VER
#include <malloc.h>
#else
#include <unistd.h>
#endif

#include <curlpp/cURLpp.hpp>
#include <deque>
#include <filesystem>
#include <fstream>
#include <future>
#include <iostream>
#include <list>
#include <memory>
#include <string>
#include <system_error>
#include <type_traits>
#include <vector>

#include "miniocpp/args.h"
#include "miniocpp/baseclient.h"
#include "miniocpp/error.h"
#include "miniocpp/http.h"
#include "miniocpp/providers.h"
#include "miniocpp/request.h"
#include "miniocpp/response.h"
#include "miniocpp/sse.h"
#include "miniocpp/types.h"
#include "miniocpp/utils.h"

#ifdef MINIO_CPP_RDMA
#include "miniocpp/nvidia-cuobjclient.h"
#include "miniocpp/rdma.h"
#endif

namespace minio::s3 {

namespace {

#ifdef _MSC_VER
inline size_t GetPageSize() { return 4096; }
inline int AlignedAlloc(void** out, size_t alignment, size_t size) {
  *out = _aligned_malloc(size, alignment);
  return *out ? 0 : -1;
}
inline void AlignedFree(void* p) { _aligned_free(p); }
#else
inline size_t GetPageSize() { return static_cast<size_t>(getpagesize()); }
inline int AlignedAlloc(void** out, size_t alignment, size_t size) {
  return posix_memalign(out, alignment, size);
}
inline void AlignedFree(void* p) { free(p); }
#endif

struct AlignedBuffer {
  void* ptr = nullptr;
  AlignedBuffer() = default;
  explicit AlignedBuffer(void* p) : ptr(p) {}
  AlignedBuffer(const AlignedBuffer&) = delete;
  AlignedBuffer& operator=(const AlignedBuffer&) = delete;
  AlignedBuffer(AlignedBuffer&& o) noexcept : ptr(o.ptr) { o.ptr = nullptr; }
  AlignedBuffer& operator=(AlignedBuffer&& o) noexcept {
    if (this != &o) {
      if (ptr) AlignedFree(ptr);
      ptr = o.ptr;
      o.ptr = nullptr;
    }
    return *this;
  }
  ~AlignedBuffer() {
    if (ptr) AlignedFree(ptr);
  }
};

#ifdef MINIO_CPP_RDMA
// Releases an RDMA buffer registration when it goes out of scope. Declared
// *after* the buffer it covers, so destruction order (reverse of declaration)
// guarantees deregister-before-free regardless of which control path returns.
struct ScopedRDMARegistration {
  cuObjClient* client = nullptr;
  void* buf = nullptr;
  ScopedRDMARegistration() = default;
  ScopedRDMARegistration(cuObjClient* c, void* p) : client(c), buf(p) {}
  ScopedRDMARegistration(const ScopedRDMARegistration&) = delete;
  ScopedRDMARegistration& operator=(const ScopedRDMARegistration&) = delete;
  ScopedRDMARegistration(ScopedRDMARegistration&& o) noexcept
      : client(o.client), buf(o.buf) {
    o.client = nullptr;
    o.buf = nullptr;
  }
  ScopedRDMARegistration& operator=(ScopedRDMARegistration&& o) noexcept {
    if (this != &o) {
      Release();
      client = o.client;
      buf = o.buf;
      o.client = nullptr;
      o.buf = nullptr;
    }
    return *this;
  }
  ~ScopedRDMARegistration() { Release(); }

 private:
  void Release() {
    if (client && buf && client->cuMemObjPutDescriptor(buf) != 0) {
      std::cerr << "warning: cuMemObjPutDescriptor failed during teardown"
                << std::endl;
    }
    client = nullptr;
    buf = nullptr;
  }
};
#endif

}  // namespace

ListObjectsResult::ListObjectsResult(error::Error err) : failed_(true) {
  resp_ = std::make_shared<ListObjectsResponse>();
  resp_->contents.push_back(Item(std::move(err)));
  itr_ = resp_->contents.begin();
}

ListObjectsResult::ListObjectsResult(Client* const client,
                                     const ListObjectsArgs& args)
    : client_(client), args_(args) {
  resp_ = std::make_shared<ListObjectsResponse>();
  itr_ = resp_->contents.end();
  StartPrefetch();
}

ListObjectsResult::ListObjectsResult(Client* const client,
                                     ListObjectsArgs&& args)
    : client_(client), args_(std::move(args)) {
  resp_ = std::make_shared<ListObjectsResponse>();
  itr_ = resp_->contents.end();
  StartPrefetch();
}

void ListObjectsResult::UpdatePaginationArgs() {
  if (args_.include_versions || !args_.version_id_marker.empty()) {
    args_.key_marker = resp_->next_key_marker;
    args_.version_id_marker = resp_->next_version_id_marker;
  } else if (args_.use_api_v1) {
    args_.marker = resp_->next_marker;
  } else {
    args_.start_after = resp_->start_after;
    args_.continuation_token = resp_->next_continuation_token;
  }
}

void ListObjectsResult::StartPrefetch() {
  ListObjectsArgs next_args = args_;
  try {
    prefetch_future_ = std::make_shared<
        std::shared_future<std::shared_ptr<ListObjectsResponse>>>(std::async(
        std::launch::async,
        [client = client_, next_args = std::move(next_args)]() mutable
            -> std::shared_ptr<ListObjectsResponse> {
          try {
            GetRegionResponse resp =
                client->GetRegion(next_args.bucket, next_args.region);
            if (resp) {
              next_args.region = resp.region;
              if (next_args.recursive) {
                next_args.delimiter = "";
              } else if (next_args.delimiter.empty()) {
                next_args.delimiter = "/";
              }

              if (next_args.include_versions ||
                  !next_args.version_id_marker.empty()) {
                return std::make_shared<ListObjectsResponse>(
                    client->ListObjectVersions(
                        ListObjectVersionsArgs(next_args)));
              } else if (next_args.use_api_v1) {
                return std::make_shared<ListObjectsResponse>(
                    client->ListObjectsV1(ListObjectsV1Args(next_args)));
              } else {
                return std::make_shared<ListObjectsResponse>(
                    client->ListObjectsV2(ListObjectsV2Args(next_args)));
              }
            }
            return std::make_shared<ListObjectsResponse>(resp);
          } catch (const std::exception& e) {
            return std::make_shared<ListObjectsResponse>(
                error::Error(std::string("prefetch failed: ") + e.what()));
          }
        }));
  } catch (const std::exception& e) {
    std::promise<std::shared_ptr<ListObjectsResponse>> p;
    p.set_value(std::make_shared<ListObjectsResponse>(
        error::Error(std::string("failed to launch prefetch: ") + e.what())));
    prefetch_future_ = std::make_shared<
        std::shared_future<std::shared_ptr<ListObjectsResponse>>>(
        p.get_future());
  }
}

void ListObjectsResult::Populate() {
  if (!prefetch_future_ || !prefetch_future_->valid()) {
    return;
  }
  try {
    resp_ = prefetch_future_->get();
  } catch (const std::exception& e) {
    resp_ = std::make_shared<ListObjectsResponse>(
        error::Error(std::string("prefetch result failed: ") + e.what()));
  }
  prefetch_future_.reset();
  if (!*resp_) {
    failed_ = true;
    resp_->contents.push_back(Item(*resp_));
  }
  itr_ = resp_->contents.begin();

  if (*resp_ && resp_->is_truncated) {
    UpdatePaginationArgs();
    StartPrefetch();
  }
}

RemoveObjectsResult::RemoveObjectsResult(error::Error err) {
  done_ = true;
  resp_.errors.push_back(DeleteError(err));
  itr_ = resp_.errors.begin();
}

RemoveObjectsResult::RemoveObjectsResult(Client* const client,
                                         const RemoveObjectsArgs& args)
    : client_(client), args_(args) {
  Populate();
}

RemoveObjectsResult::RemoveObjectsResult(Client* const client,
                                         RemoveObjectsArgs&& args)
    : client_(client), args_(args) {
  Populate();
}

void RemoveObjectsResult::Populate() {
  while (!done_ && resp_.errors.size() == 0) {
    RemoveObjectsApiArgs args;
    args.extra_headers = args_.extra_headers;
    args.extra_query_params = args_.extra_query_params;
    args.bucket = args_.bucket;
    args.region = args_.region;
    args.quiet = true;
    args.bypass_governance_mode = args_.bypass_governance_mode;

    for (int i = 0; i < 1000; i++) {
      DeleteObject object;
      if (!args_.func(object)) {
        break;
      }
      args.objects.push_back(object);
    }

    if (args.objects.size() != 0) {
      resp_ = client_->BaseClient::RemoveObjects(args);
      if (!resp_) {
        resp_.errors.push_back(DeleteError(resp_));
      }
      itr_ = resp_.errors.begin();
    } else {
      done_ = true;
    }
  }
  // Caller's func may have returned false on the very first call (empty
  // batch). `done_` flips to true above but itr_ was never assigned, so
  // operator bool() would compare an uninitialized iterator. Pin it to
  // end() so the result evaluates to false cleanly.
  if (done_ && resp_.errors.empty()) {
    itr_ = resp_.errors.end();
  }
}

#ifdef MINIO_CPP_RDMA
// Meyers singleton — thread-safe per C++11 [stmt.dcl]/4 ("If control
// enters the declaration concurrently while the variable is being
// initialized, the concurrent execution shall wait for completion of the
// initialization."). This replaces the previous per-call cuObjClient
// construction, which was racy under concurrency and caused the
// glibc-level "malloc(): invalid size (unsorted)" abort when multiple
// warp workers started up simultaneously.
cuObjClient& Client::SharedRDMAClient() {
  static CUObjIOOps ops{};
  static cuObjClient client{ops, CUOBJ_PROTO_RDMA_DC_V1};
  return client;
}
#endif

Client::Client(BaseUrl& base_url, creds::Provider* const provider)
    : BaseClient(base_url, provider) {}

StatObjectResponse Client::CalculatePartCount(
    size_t& part_count, std::list<ComposeSource> sources) {
  size_t object_size = 0;
  size_t i = 0;
  for (auto& source : sources) {
    if (source.ssec != nullptr && !base_url_.https) {
      std::string msg = "source " + source.bucket + "/" + source.object;
      if (!source.version_id.empty()) {
        msg += "?versionId=" + source.version_id;
      }
      msg += ": SSE-C operation must be performed over a secure connection";
      return error::make<StatObjectResponse>(msg);
    }

    i++;

    std::string etag;
    size_t size;

    StatObjectResponse resp = StatObject(source);
    if (!resp) {
      return resp;
    }
    etag = resp.etag;
    size = resp.size;
    if (error::Error err = source.BuildHeaders(size, etag)) {
      return StatObjectResponse(err);
    }
    if (source.length.has_value()) {
      size = *source.length;
    } else if (source.offset.has_value()) {
      size -= *source.offset;
    }

    if (size < utils::kMinPartSize && sources.size() != 1 &&
        i != sources.size()) {
      std::string msg = "source " + source.bucket + "/" + source.object;
      if (!source.version_id.empty()) msg += "?versionId=" + source.version_id;
      msg += ": size " + std::to_string(size) + " must be greater than " +
             std::to_string(utils::kMinPartSize);
      return error::make<StatObjectResponse>(msg);
    }

    object_size += size;
    if (object_size > utils::kMaxObjectSize) {
      return error::make<StatObjectResponse>(
          "destination object size must be less than " +
          std::to_string(utils::kMaxObjectSize));
    }

    if (size > utils::kMaxPartSize) {
      size_t count = size / utils::kMaxPartSize;
      size_t last_part_size = size - (count * utils::kMaxPartSize);
      if (last_part_size > 0) {
        count++;
      } else {
        last_part_size = utils::kMaxPartSize;
      }

      if (last_part_size < utils::kMinPartSize && sources.size() != 1 &&
          i != sources.size()) {
        std::string msg = "source " + source.bucket + "/" + source.object;
        if (!source.version_id.empty()) {
          msg += "?versionId=" + source.version_id;
        }
        msg += ": size " + std::to_string(size) +
               " for multipart split upload of " + std::to_string(size) +
               ", last part size is less than " +
               std::to_string(utils::kMinPartSize);
        return error::make<StatObjectResponse>(msg);
      }

      part_count += count;
    } else {
      part_count++;
    }

    if (part_count > utils::kMaxMultipartCount) {
      return error::make<StatObjectResponse>(
          "Compose sources create more than allowed multipart count " +
          std::to_string(utils::kMaxMultipartCount));
    }
  }

  return StatObjectResponse(error::SUCCESS);
}

ComposeObjectResponse Client::ComposeObject(ComposeObjectArgs args,
                                            std::string& upload_id) {
  size_t part_count = 0;
  {
    StatObjectResponse resp = CalculatePartCount(part_count, args.sources);
    if (!resp) {
      return ComposeObjectResponse(resp);
    }
  }

  ComposeSource& source = args.sources.front();
  if (part_count == 1 && !source.offset.has_value() &&
      !source.length.has_value()) {
    CopyObjectArgs coargs;
    coargs.extra_headers = args.extra_headers;
    coargs.extra_query_params = args.extra_query_params;
    coargs.bucket = args.bucket;
    coargs.region = args.region;
    coargs.object = args.object;
    coargs.sse = args.sse;
    coargs.source = source;

    return ComposeObjectResponse(CopyObject(coargs));
  }

  utils::Multimap headers = args.Headers();

  {
    CreateMultipartUploadArgs cmu_args;
    cmu_args.extra_query_params = args.extra_query_params;
    cmu_args.bucket = args.bucket;
    cmu_args.region = args.region;
    cmu_args.object = args.object;
    cmu_args.headers = headers;
    if (CreateMultipartUploadResponse resp = CreateMultipartUpload(cmu_args)) {
      upload_id = resp.upload_id;
    } else {
      return ComposeObjectResponse(resp);
    }
  }

  unsigned int part_number = 0;
  utils::Multimap ssecheaders;
  if (args.sse != nullptr) {
    if (SseCustomerKey* ssec = dynamic_cast<SseCustomerKey*>(args.sse)) {
      ssecheaders = ssec->Headers();
    }
  }

  std::list<Part> parts;
  for (auto& source : args.sources) {
    size_t size = source.ObjectSize();
    if (source.length.has_value()) {
      size = *source.length;
    } else if (source.offset.has_value()) {
      size -= *source.offset;
    }

    size_t offset = 0;
    if (source.offset.has_value()) offset = *source.offset;

    utils::Multimap headers;
    headers.AddAll(source.Headers());
    headers.AddAll(ssecheaders);

    if (size <= utils::kMaxPartSize) {
      part_number++;
      if (source.length.has_value()) {
        headers.Add("x-amz-copy-source-range",
                    "bytes=" + std::to_string(offset) + "-" +
                        std::to_string(offset + *source.length - 1));
      } else if (source.offset.has_value()) {
        headers.Add("x-amz-copy-source-range",
                    "bytes=" + std::to_string(offset) + "-" +
                        std::to_string(offset + size - 1));
      }

      UploadPartCopyArgs upc_args;
      upc_args.bucket = args.bucket;
      upc_args.region = args.region;
      upc_args.object = args.object;
      upc_args.headers = headers;
      upc_args.upload_id = upload_id;
      upc_args.part_number = part_number;
      UploadPartCopyResponse resp = UploadPartCopy(upc_args);
      if (!resp) {
        return ComposeObjectResponse(resp);
      }
      parts.push_back(Part(part_number, std::move(resp.etag)));
    } else {
      while (size > 0) {
        part_number++;

        size_t length = size;
        if (length > utils::kMaxPartSize) length = utils::kMaxPartSize;
        size_t end_bytes = offset + length - 1;

        utils::Multimap headerscopy;
        headerscopy.AddAll(headers);
        headerscopy.Add("x-amz-copy-source-range",
                        "bytes=" + std::to_string(offset) + "-" +
                            std::to_string(end_bytes));

        UploadPartCopyArgs upc_args;
        upc_args.bucket = args.bucket;
        upc_args.region = args.region;
        upc_args.object = args.object;
        upc_args.headers = headerscopy;
        upc_args.upload_id = upload_id;
        upc_args.part_number = part_number;
        {
          UploadPartCopyResponse resp = UploadPartCopy(upc_args);
          if (!resp) {
            return ComposeObjectResponse(resp);
          }
          parts.push_back(Part(part_number, std::move(resp.etag)));
        }
        offset += length;
        size -= length;
      }
    }
  }

  CompleteMultipartUploadArgs cmu_args;
  cmu_args.bucket = args.bucket;
  cmu_args.region = args.region;
  cmu_args.object = args.object;
  cmu_args.upload_id = upload_id;
  cmu_args.parts = parts;
  return ComposeObjectResponse(CompleteMultipartUpload(cmu_args));
}

GetObjectResponse Client::GetObject(GetObjectArgs args) {
  if (error::Error err = args.Validate()) {
    return GetObjectResponse(err);
  }

#ifdef MINIO_CPP_RDMA
  if (args.buf != nullptr) {
    std::string region;
    if (GetRegionResponse resp = GetRegion(args.bucket, args.region)) {
      region = resp.region;
    } else {
      return GetObjectResponse(resp);
    }

    const size_t size = *args.size;

    // Process-wide cuObjClient — see client.h for the race rationale.
    cuObjClient& rdma_client = SharedRDMAClient();
    bool use_rdma = (rdma_client.cuMemObjGetDescriptor(args.buf, size) == 0);

    if (use_rdma) {
      s3_rdma_client_ctx getCtx = {
          .provider = provider_,
          .bucket = args.bucket,
          .object = args.object,
          .url = base_url_,
          .region = region,
          .op = CUOBJ_GET,
      };

      ssize_t ret = rdmaGetWithRetry(&rdma_client, &getCtx, args.buf, size);
      rdma_client.cuMemObjPutDescriptor(args.buf);

      if (ret > 0) {
        GetObjectResponse resp;
        resp.etag = getCtx.etag;
        return resp;
      }
      // ret < 0 (retries exhausted) or kRDMANotSupported (server declined):
      // fall through to HTTP-into-buffer path below.
    }

    // HTTP fallback: stream the body into the caller's buffer.
    GetObjectArgs targs;
    std::stringstream ss(std::ios_base::in | std::ios_base::out);
    ss.rdbuf()->pubsetbuf(args.buf, size);

    targs.bucket = args.bucket;
    targs.object = args.object;
    targs.region = region;
    targs.datafunc = [&ss = ss](minio::http::DataFunctionArgs args) -> bool {
      ss << args.datachunk;
      return true;
    };

    return BaseClient::GetObject(targs);
  }
#endif

  return BaseClient::GetObject(args);
}

PutObjectResponse Client::PutObject(PutObjectArgs args, std::string& upload_id,
                                    char* buf) {
  utils::Multimap headers = args.Headers();
  if (!headers.Contains("Content-Type")) {
    if (args.content_type.empty()) {
      headers.Add("Content-Type", "application/octet-stream");
    } else {
      headers.Add("Content-Type", args.content_type);
    }
  }

  std::optional<uint64_t> object_size = args.object_size;
  size_t part_size = args.part_size;
  size_t uploaded_size = 0;
  unsigned int part_number = 0;
  std::string one_byte;
  bool stop = false;
  std::list<Part> parts;
  std::optional<size_t> part_count = args.part_count;
  double uploaded_bytes = 0;           // for progress
  std::optional<double> upload_speed;  // for progress

  auto read_part_data = [&](char* buf, size_t& bytes_read) -> error::Error {
    if (part_count.has_value()) {
      if (part_number == *part_count) {
        part_size = *object_size - uploaded_size;
        stop = true;
      }

      if (error::Error err =
              utils::ReadPart(*args.stream, buf, part_size, bytes_read)) {
        return err;
      }

      if (bytes_read != part_size) {
        return error::Error("not enough data in the stream; expected: " +
                            std::to_string(part_size) +
                            ", got: " + std::to_string(bytes_read) + " bytes");
      }
    } else {
      char* b = buf;
      size_t size = part_size + 1;

      if (!one_byte.empty()) {
        buf[0] = one_byte.front();
        b = buf + 1;
        size--;
        bytes_read = 1;
        one_byte = "";
      }

      size_t n = 0;
      if (error::Error err = utils::ReadPart(*args.stream, b, size, n)) {
        return err;
      }

      bytes_read += n;

      // If bytes read is less than or equals to part size, then we have reached
      // last part.
      if (bytes_read <= part_size) {
        part_count = std::optional<size_t>(part_number);
        part_size = bytes_read;
        stop = true;
      } else {
        one_byte = buf[part_size];
      }
    }
    return error::Error();
  };

  while (!stop) {
    part_number++;

    size_t bytes_read = 0;
    if (error::Error err = read_part_data(buf, bytes_read)) {
      return PutObjectResponse(err);
    }

    std::string_view data(buf, part_size);

    uploaded_size += part_size;

    if (part_count.has_value() && *part_count == 1) {
      PutObjectApiArgs api_args;
      api_args.extra_query_params = args.extra_query_params;
      api_args.bucket = args.bucket;
      api_args.region = args.region;
      api_args.object = args.object;
      api_args.data = data;
      api_args.buf = buf;
      api_args.size = part_size;
      api_args.progressfunc = args.progressfunc;
      api_args.progress_userdata = args.progress_userdata;
      api_args.headers = headers;

      return BaseClient::PutObject(api_args);
    }

    if (upload_id.empty()) {
      CreateMultipartUploadArgs cmu_args;
      cmu_args.extra_query_params = args.extra_query_params;
      cmu_args.bucket = args.bucket;
      cmu_args.region = args.region;
      cmu_args.object = args.object;
      cmu_args.headers = headers;
#ifdef MINIO_CPP_RDMA
      // Declare CRC64NVME so the server enforces per-part integrity on the
      // RDMA UploadPart path (server returns 501 if checksum is missing when
      // an algorithm was declared on Create).
      cmu_args.headers.Add("x-amz-checksum-algorithm", "CRC64NVME");
#endif
      if (CreateMultipartUploadResponse resp =
              CreateMultipartUpload(cmu_args)) {
        upload_id = resp.upload_id;
      } else {
        return PutObjectResponse(resp);
      }
    }

    UploadPartArgs up_args;
    up_args.bucket = args.bucket;
    up_args.region = args.region;
    up_args.object = args.object;
    up_args.upload_id = upload_id;
    up_args.part_number = part_number;
    up_args.data = data;
    up_args.buf = buf;
    up_args.part_size = part_size;
#ifdef MINIO_CPP_RDMA
    up_args.rdmaclient = args.rdmaclient;
    if (buf != nullptr &&
        cuObjClient::getMemoryType(buf) == CUOBJ_MEMORY_SYSTEM) {
      const std::string crc = utils::Crc64NvmeBase64(buf, part_size);
      up_args.checksum_crc64nvme = crc;
      up_args.headers.Add("x-amz-checksum-crc64nvme", crc);
    }
#endif
    if (args.progressfunc != nullptr) {
      up_args.progressfunc =
          [&object_size = object_size, &uploaded_bytes = uploaded_bytes,
           &upload_speed = upload_speed, &progressfunc = args.progressfunc,
           &progress_userdata = args.progress_userdata](
              http::ProgressFunctionArgs args) -> bool {
        if (args.upload_speed > 0) {
          if (!upload_speed.has_value()) {
            upload_speed = args.upload_speed;
          } else {
            upload_speed = (*upload_speed + args.upload_speed) / 2.0;
          }
          return true;
        }

        http::ProgressFunctionArgs actual_args;
        actual_args.upload_total_bytes =
            object_size ? static_cast<double>(*object_size) : -1.0;
        actual_args.uploaded_bytes = uploaded_bytes + args.uploaded_bytes;
        actual_args.userdata = progress_userdata;
        return progressfunc(actual_args);
      };
    }
    // Propagate caller-supplied x-amz-content-sha256 (e.g. UNSIGNED-PAYLOAD
    // for GPU-resident buffers) into each UploadPart so the per-part signing
    // path also skips hashing the body.
    if (headers.Contains("x-amz-content-sha256")) {
      up_args.headers.Add("x-amz-content-sha256",
                          headers.GetFront("x-amz-content-sha256"));
    }
    if (args.sse != nullptr) {
      if (SseCustomerKey* ssec = dynamic_cast<SseCustomerKey*>(args.sse)) {
        up_args.headers.AddAll(ssec->Headers());
      }
    }

    if (UploadPartResponse resp = UploadPart(up_args)) {
      if (args.progressfunc != nullptr) {
        uploaded_bytes += static_cast<double>(data.length());
        http::ProgressFunctionArgs actual_args;
        actual_args.upload_total_bytes =
            object_size ? static_cast<double>(*object_size) : -1.0;
        actual_args.uploaded_bytes = uploaded_bytes;
        actual_args.userdata = args.progress_userdata;
        if (!args.progressfunc(actual_args)) {
          return UploadPartResponse(
              error::Error("aborted by progress function"));
        }
      }
      // HTTP fallback leaves resp.checksum_crc64nvme empty; use the local CRC.
      parts.push_back(Part(part_number, std::move(resp.etag),
                           std::move(up_args.checksum_crc64nvme)));
    } else {
      return resp;
    }
  }

  CompleteMultipartUploadArgs cmu_args;
  cmu_args.bucket = args.bucket;
  cmu_args.region = args.region;
  cmu_args.object = args.object;
  cmu_args.upload_id = upload_id;
  cmu_args.parts = parts;
  CompleteMultipartUploadResponse resp = CompleteMultipartUpload(cmu_args);
  if (resp && args.progressfunc != nullptr) {
    http::ProgressFunctionArgs actual_args;
    actual_args.upload_speed = upload_speed.value_or(-1.0);
    actual_args.userdata = args.progress_userdata;
    // ignore the return value as we completed the upload
    args.progressfunc(actual_args);
  }
  return PutObjectResponse(resp);
}

ComposeObjectResponse Client::ComposeObject(ComposeObjectArgs args) {
  if (error::Error err = args.Validate()) {
    return ComposeObjectResponse(err);
  }

  if (args.sse != nullptr && args.sse->TlsRequired() && !base_url_.https) {
    return error::make<ComposeObjectResponse>(
        "SSE operation must be performed over a secure connection");
  }

  std::string upload_id;
  ComposeObjectResponse resp = ComposeObject(args, upload_id);
  if (!resp && !upload_id.empty()) {
    AbortMultipartUploadArgs amu_args;
    amu_args.bucket = args.bucket;
    amu_args.region = args.region;
    amu_args.object = args.object;
    amu_args.upload_id = upload_id;
    AbortMultipartUpload(amu_args);
  }

  return resp;
}

CopyObjectResponse Client::CopyObject(CopyObjectArgs args) {
  if (error::Error err = args.Validate()) {
    return CopyObjectResponse(err);
  }

  if (args.sse != nullptr && args.sse->TlsRequired() && !base_url_.https) {
    return error::make<CopyObjectResponse>(
        "SSE operation must be performed over a secure connection");
  }

  if (args.source.ssec != nullptr && !base_url_.https) {
    return error::make<CopyObjectResponse>(
        "SSE-C operation must be performed over a secure connection");
  }

  std::string etag;
  size_t size;
  {
    StatObjectResponse resp = StatObject(args.source);
    if (!resp) {
      return CopyObjectResponse(resp);
    }
    etag = resp.etag;
    size = resp.size;
  }

  if (args.source.offset.has_value() || args.source.length.has_value() ||
      size > utils::kMaxPartSize) {
    if (args.metadata_directive != nullptr &&
        *args.metadata_directive == Directive::kCopy) {
      return error::make<CopyObjectResponse>(
          "COPY metadata directive is not applicable to source object size "
          "greater than 5 GiB");
    }

    if (args.tagging_directive != nullptr &&
        *args.tagging_directive == Directive::kCopy) {
      return error::make<CopyObjectResponse>(
          "COPY tagging directive is not applicable to source object size "
          "greater than 5 GiB");
    }

    ComposeSource src;
    src.extra_headers = args.source.extra_headers;
    src.extra_query_params = args.source.extra_query_params;
    src.bucket = args.source.bucket;
    src.region = args.source.region;
    src.object = args.source.object;
    src.ssec = args.source.ssec;
    src.offset = args.source.offset;
    src.length = args.source.length;
    src.match_etag = args.source.match_etag;
    src.not_match_etag = args.source.not_match_etag;
    src.modified_since = args.source.modified_since;
    src.unmodified_since = args.source.unmodified_since;

    ComposeObjectArgs coargs;
    coargs.extra_headers = args.extra_headers;
    coargs.extra_query_params = args.extra_query_params;
    coargs.bucket = args.bucket;
    coargs.region = args.region;
    coargs.object = args.object;
    coargs.sse = args.sse;
    coargs.sources.push_back(src);

    return CopyObjectResponse(ComposeObject(coargs));
  }

  utils::Multimap headers;
  headers.AddAll(args.extra_headers);
  headers.AddAll(args.Headers());
  if (args.metadata_directive != nullptr) {
    headers.Add("x-amz-metadata-directive",
                DirectiveToString(*args.metadata_directive));
  }
  if (args.tagging_directive != nullptr) {
    headers.Add("x-amz-tagging-directive",
                DirectiveToString(*args.tagging_directive));
  }
  headers.AddAll(args.source.CopyHeaders());

  std::string region;
  if (GetRegionResponse resp = GetRegion(args.bucket, args.region)) {
    region = resp.region;
  } else {
    return CopyObjectResponse(resp);
  }

  Request req(http::Method::kPut, region, base_url_, args.extra_headers,
              args.extra_query_params);
  req.bucket_name = args.bucket;
  req.object_name = args.object;
  req.headers.AddAll(headers);

  Response response = Execute(req);
  if (!response) {
    return CopyObjectResponse(response);
  }

  CopyObjectResponse resp;
  resp.etag = utils::Trim(response.headers.GetFront("etag"), '"');
  resp.version_id = response.headers.GetFront("x-amz-version-id");

  return resp;
}

DownloadObjectResponse Client::DownloadObject(DownloadObjectArgs args) {
  if (error::Error err = args.Validate()) {
    return DownloadObjectResponse(err);
  }

  if (args.ssec != nullptr && !base_url_.https) {
    return error::make<DownloadObjectResponse>(
        "SSE-C operation must be performed over a secure connection");
  }

  std::string etag;
  {
    StatObjectArgs soargs;
    soargs.bucket = args.bucket;
    soargs.region = args.region;
    soargs.object = args.object;
    soargs.version_id = args.version_id;
    soargs.ssec = args.ssec;
    StatObjectResponse resp = StatObject(soargs);
    if (!resp) {
      return DownloadObjectResponse(resp);
    }
    etag = resp.etag;
  }

  std::string temp_filename =
      args.filename + "." + curlpp::escape(etag) + ".part.minio";
  std::ofstream fout(temp_filename, std::ios::trunc | std::ios::out);
  if (!fout.is_open()) {
    return error::make<DownloadObjectResponse>("unable to open file " +
                                               temp_filename);
  }

  std::string region;
  if (GetRegionResponse resp = GetRegion(args.bucket, args.region)) {
    region = resp.region;
  } else {
    return DownloadObjectResponse(resp);
  }

  Request req(http::Method::kGet, region, base_url_, args.extra_headers,
              args.extra_query_params);
  req.bucket_name = args.bucket;
  req.object_name = args.object;
  if (!args.version_id.empty()) {
    req.query_params.Add("versionId", args.version_id);
  }
  req.datafunc = [&fout = fout](http::DataFunctionArgs args) -> bool {
    fout << args.datachunk;
    return true;
  };
  req.progressfunc = args.progressfunc;
  req.progress_userdata = args.progress_userdata;

  Response response = Execute(req);
  fout.close();
  if (response) {
    std::filesystem::rename(temp_filename, args.filename);
  }
  return DownloadObjectResponse(response);
}

ListObjectsResult Client::ListObjects(ListObjectsArgs args) {
  if (error::Error err = args.Validate()) {
    return ListObjectsResult(err);
  }
  return ListObjectsResult(this, std::move(args));
}

PutObjectResponse Client::PutObject(PutObjectArgs args) {
  if (error::Error err = args.Validate()) {
    return PutObjectResponse(err);
  }

  if (args.sse != nullptr && args.sse->TlsRequired() && !base_url_.https) {
    return error::make<PutObjectResponse>(
        "SSE operation must be performed over a secure connection");
  }

#ifdef MINIO_CPP_RDMA
  // Direct-buffer mode (was PutObject(PutObjectRDMAArgs)): caller supplied
  // a pre-allocated, page-aligned buffer. Try RDMA, fall back to a single
  // HTTP upload from the same buffer if RDMA declines.
  if (args.buf != nullptr) {
    std::string region;
    if (GetRegionResponse resp = GetRegion(args.bucket, args.region)) {
      region = resp.region;
    } else {
      return PutObjectResponse(resp);
    }

    const size_t size = *args.size;

    cuObjClient& rdma_client = SharedRDMAClient();
    bool use_rdma = (rdma_client.cuMemObjGetDescriptor(args.buf, size) == 0);

    if (use_rdma) {
      s3_rdma_client_ctx putCtx = {
          .provider = provider_,
          .bucket = args.bucket,
          .object = args.object,
          .url = base_url_,
          .region = region,
          .op = CUOBJ_PUT,
      };

      ssize_t ret = rdmaPutWithRetry(&rdma_client, &putCtx, args.buf, size);
      rdma_client.cuMemObjPutDescriptor(args.buf);

      if (ret > 0) {
        PutObjectResponse resp;
        resp.etag = putCtx.etag;
        return resp;
      }
      // ret < 0 / kRDMANotSupported: fall through to HTTP-from-buffer.
    }

    // HTTP fallback from the same buffer (single-shot, not multipart).
    // Skip body hashing for signing — caller's buffer may be GPU-resident,
    // and we'd otherwise drag device memory through OpenSSL just to compute
    // a hash that TLS already authenticates.
    std::stringstream ss(std::ios_base::in | std::ios_base::out);
    ss.rdbuf()->pubsetbuf(args.buf, size);

    PutObjectArgs http_args(ss, static_cast<uint64_t>(size), 16 * 1024 * 1024L);
    http_args.bucket = args.bucket;
    http_args.object = args.object;
    http_args.region = region;
    http_args.headers.Add("x-amz-content-sha256", "UNSIGNED-PAYLOAD");
    return PutObject(http_args);
  }
#endif

  if (args.part_count.has_value() && *args.part_count == 1) {
    args.max_inflight_parts = 1;
  }

  // === Parallel multipart upload with bounded inflight ===
  unsigned int max_inflight = args.max_inflight_parts.value_or(1);
  if (max_inflight > 1) {
    // Clamp to a reasonable maximum and to part_count to prevent memory
    // exhaustion from untrusted input.
    constexpr unsigned int kMaxInflightParts = 100;
    if (max_inflight > kMaxInflightParts) {
      max_inflight = kMaxInflightParts;
    }
    if (args.part_count.has_value() && *args.part_count > 0 &&
        static_cast<size_t>(max_inflight) > *args.part_count) {
      max_inflight = static_cast<unsigned int>(*args.part_count);
    }

    size_t alloc_size =
        (args.part_count > 0) ? args.part_size : (args.part_size + 1);

    // Allocate pool of page-aligned buffers.
    std::vector<AlignedBuffer> buf_pool;
    for (unsigned int i = 0; i < max_inflight; i++) {
      void* raw = nullptr;
      if (AlignedAlloc(&raw, GetPageSize(), alloc_size)) {
        return error::make<PutObjectResponse>(
            "unable to allocate aligned buffer");
      }
      buf_pool.emplace_back(raw);
    }

#ifdef MINIO_CPP_RDMA
    // Register each pool buffer for RDMA, indexed by buffer slot.
    cuObjClient& rdma_client = SharedRDMAClient();
    std::vector<ScopedRDMARegistration> rdma_regs(max_inflight);
    bool rdma_connected = rdma_client.isConnected();
    if (rdma_connected) {
      for (unsigned int i = 0; i < max_inflight; i++) {
        char* pool_buf = static_cast<char*>(buf_pool[i].ptr);
        if (rdma_client.cuMemObjGetDescriptor(pool_buf, args.part_size) == 0) {
          rdma_regs[i] = ScopedRDMARegistration(&rdma_client, pool_buf);
        }
      }
    }
#endif

    utils::Multimap headers = args.Headers();
    if (!headers.Contains("Content-Type")) {
      if (args.content_type.empty()) {
        headers.Add("Content-Type", "application/octet-stream");
      } else {
        headers.Add("Content-Type", args.content_type);
      }
    }

    std::optional<uint64_t> object_size = args.object_size;
    size_t part_size = args.part_size;
    size_t uploaded_size = 0;
    unsigned int part_number = 0;
    std::string one_byte;
    bool stop = false;
    std::list<Part> parts;
    std::optional<size_t> part_count = args.part_count;
    std::string upload_id;
    error::Error first_err;
    double uploaded_bytes = 0;           // for progress
    std::optional<double> upload_speed;  // for progress

    struct InflightPart {
      unsigned int part_number;
      std::string checksum_crc64nvme;
      size_t part_bytes;
      std::future<UploadPartResponse> future;
    };
    std::deque<InflightPart> inflight;

    unsigned int buf_idx = 0;

    auto report_progress = [&](size_t part_bytes) -> bool {
      if (args.progressfunc == nullptr) return true;
      uploaded_bytes += static_cast<double>(part_bytes);
      http::ProgressFunctionArgs actual_args;
      actual_args.upload_total_bytes =
          object_size ? static_cast<double>(*object_size) : -1.0;
      actual_args.uploaded_bytes = uploaded_bytes;
      actual_args.userdata = args.progress_userdata;
      if (!args.progressfunc(actual_args)) {
        first_err = error::Error("aborted by progress function");
        return false;
      }
      return true;
    };

    auto read_part_data = [&](char* buf, size_t& bytes_read) -> error::Error {
      if (part_count.has_value()) {
        if (part_number == *part_count) {
          part_size = *object_size - uploaded_size;
          stop = true;
        }

        if (error::Error err =
                utils::ReadPart(*args.stream, buf, part_size, bytes_read)) {
          return err;
        }

        if (bytes_read != part_size) {
          return error::Error("not enough data in the stream; expected: " +
                              std::to_string(part_size) + ", got: " +
                              std::to_string(bytes_read) + " bytes");
        }
      } else {
        char* b = buf;
        size_t size = part_size + 1;

        if (!one_byte.empty()) {
          buf[0] = one_byte.front();
          b = buf + 1;
          size--;
          bytes_read = 1;
          one_byte = "";
        }

        size_t n = 0;
        if (error::Error err = utils::ReadPart(*args.stream, b, size, n)) {
          return err;
        }

        bytes_read += n;

        // If bytes read is less than or equals to part size, then we have
        // reached last part.
        if (bytes_read <= part_size) {
          part_count = std::optional<size_t>(part_number);
          part_size = bytes_read;
          stop = true;
        } else {
          one_byte = buf[part_size];
        }
      }
      return error::Error();
    };

    while (!stop && !first_err) {
      part_number++;

      // Wait for a slot before reusing the buffer at buf_idx.
      if (inflight.size() >= max_inflight) {
        InflightPart ip = std::move(inflight.front());
        inflight.pop_front();
        UploadPartResponse resp;
        try {
          resp = ip.future.get();
        } catch (const std::exception& e) {
          first_err =
              error::Error(std::string("upload part failed: ") + e.what());
          break;
        }
        if (!resp) {
          first_err = resp.Error();
          break;
        }
        parts.push_back(Part(ip.part_number, std::move(resp.etag),
                             std::move(ip.checksum_crc64nvme)));
        if (!report_progress(ip.part_bytes)) {
          break;
        }
      }

      char* buf = static_cast<char*>(buf_pool[buf_idx].ptr);
      size_t bytes_read = 0;
      if (error::Error err = read_part_data(buf, bytes_read)) {
        first_err = err;
        break;
      }

      uploaded_size += part_size;

      // Single part discovered dynamically: use PutObject directly.
      if (part_count.has_value() && *part_count == 1) {
        PutObjectApiArgs api_args;
        api_args.extra_query_params = args.extra_query_params;
        api_args.bucket = args.bucket;
        api_args.region = args.region;
        api_args.object = args.object;
        api_args.data = std::string_view(buf, part_size);
        api_args.buf = buf;
        api_args.size = part_size;
        api_args.progressfunc = args.progressfunc;
        api_args.progress_userdata = args.progress_userdata;
        api_args.headers = headers;
        return BaseClient::PutObject(api_args);
      }

      // Create multipart upload on first part.
      if (upload_id.empty()) {
        CreateMultipartUploadArgs cmu_args;
        cmu_args.extra_query_params = args.extra_query_params;
        cmu_args.bucket = args.bucket;
        cmu_args.region = args.region;
        cmu_args.object = args.object;
        cmu_args.headers = headers;
#ifdef MINIO_CPP_RDMA
        cmu_args.headers.Add("x-amz-checksum-algorithm", "CRC64NVME");
#endif
        if (CreateMultipartUploadResponse resp =
                CreateMultipartUpload(cmu_args)) {
          upload_id = resp.upload_id;
        } else {
          first_err = resp.Error();
          break;
        }
      }

      // Build UploadPartArgs and dispatch via std::async.
      UploadPartArgs up_args;
      up_args.bucket = args.bucket;
      up_args.region = args.region;
      up_args.object = args.object;
      up_args.upload_id = upload_id;
      up_args.part_number = part_number;
      up_args.data = std::string_view(buf, part_size);
      up_args.buf = buf;
      up_args.part_size = part_size;
#ifdef MINIO_CPP_RDMA
      if (rdma_regs[buf_idx].client != nullptr) {
        up_args.rdmaclient = &rdma_client;
      }
      if (buf != nullptr &&
          cuObjClient::getMemoryType(buf) == CUOBJ_MEMORY_SYSTEM) {
        const std::string crc = utils::Crc64NvmeBase64(buf, part_size);
        up_args.checksum_crc64nvme = crc;
        up_args.headers.Add("x-amz-checksum-crc64nvme", crc);
      }
#endif
      if (headers.Contains("x-amz-content-sha256")) {
        up_args.headers.Add("x-amz-content-sha256",
                            headers.GetFront("x-amz-content-sha256"));
      }
      if (args.sse != nullptr) {
        if (SseCustomerKey* ssec = dynamic_cast<SseCustomerKey*>(args.sse)) {
          up_args.headers.AddAll(ssec->Headers());
        }
      }

      InflightPart ip;
      ip.part_number = part_number;
      ip.checksum_crc64nvme = up_args.checksum_crc64nvme;
      ip.part_bytes = part_size;
      try {
        ip.future = std::async(std::launch::async, [this, up_args]() {
          return UploadPart(up_args);
        });
      } catch (const std::system_error& e) {
        first_err =
            error::Error(std::string("unable to create thread: ") + e.what());
        break;
      }
      inflight.push_back(std::move(ip));

      buf_idx = (buf_idx + 1) % max_inflight;
    }

    // Drain all inflight parts.
    while (!inflight.empty()) {
      InflightPart ip = std::move(inflight.front());
      inflight.pop_front();
      UploadPartResponse resp;
      try {
        resp = ip.future.get();
      } catch (const std::exception& e) {
        if (!first_err)
          first_err =
              error::Error(std::string("upload part failed: ") + e.what());
        continue;
      }
      if (!resp) {
        if (!first_err) first_err = resp.Error();
        continue;
      }
      parts.push_back(Part(ip.part_number, std::move(resp.etag),
                           std::move(ip.checksum_crc64nvme)));
      if (!first_err) report_progress(ip.part_bytes);
    }

    if (first_err) {
      if (!upload_id.empty()) {
        AbortMultipartUploadArgs amu_args;
        amu_args.bucket = std::move(args.bucket);
        amu_args.region = std::move(args.region);
        amu_args.object = std::move(args.object);
        amu_args.upload_id = upload_id;
        AbortMultipartUpload(amu_args);
      }
      return PutObjectResponse(first_err);
    }

    // Complete multipart upload.
    CompleteMultipartUploadArgs cmu_args;
    cmu_args.bucket = args.bucket;
    cmu_args.region = args.region;
    cmu_args.object = args.object;
    cmu_args.upload_id = upload_id;
    cmu_args.parts = parts;
    CompleteMultipartUploadResponse cmu_resp =
        CompleteMultipartUpload(cmu_args);
    if (cmu_resp && args.progressfunc != nullptr) {
      http::ProgressFunctionArgs actual_args;
      actual_args.upload_speed = upload_speed.value_or(-1.0);
      actual_args.userdata = args.progress_userdata;
      args.progressfunc(actual_args);
    }
    if (!cmu_resp && !upload_id.empty()) {
      AbortMultipartUploadArgs amu_args;
      amu_args.bucket = std::move(args.bucket);
      amu_args.region = std::move(args.region);
      amu_args.object = std::move(args.object);
      amu_args.upload_id = upload_id;
      AbortMultipartUpload(amu_args);
    }
    return PutObjectResponse(cmu_resp);
  }

  void* raw = nullptr;
  if (AlignedAlloc(
          &raw, GetPageSize(),
          (args.part_count > 0) ? args.part_size : args.part_size + 1)) {
    return error::make<PutObjectResponse>(
        "unable to allocate system memory with alignment");
  }
  AlignedBuffer aligned_buf(raw);
  char* buf = static_cast<char*>(aligned_buf.ptr);

#ifdef MINIO_CPP_RDMA
  // Reuse the process-wide SharedRDMAClient() rather than constructing
  // per-call; see client.h for the race rationale. `buf` here is the
  // multipart part buffer, registered once for the whole multipart
  // upload and deregistered when scope exits.
  //
  // If the cuObj layer is connected but declines to register this buffer
  // (e.g. nvidia_peermem.ko not loaded, IB device unhealthy, GPUDirect
  // misconfigured), fall back to the plain HTTP multipart path instead of
  // failing the whole call — BaseClient::PutObject keys off args.rdmaclient
  // being non-null to even attempt the RDMA path.
  cuObjClient& rdma_client = SharedRDMAClient();
  ScopedRDMARegistration rdma_reg;
  if (rdma_client.isConnected() &&
      rdma_client.cuMemObjGetDescriptor(buf, args.part_size) == 0) {
    rdma_reg = ScopedRDMARegistration(&rdma_client, buf);
    args.rdmaclient = &rdma_client;
  }
#endif

  std::string upload_id;
  PutObjectResponse resp = PutObject(args, upload_id, buf);

  if (!resp && !upload_id.empty()) {
    AbortMultipartUploadArgs amu_args;
    amu_args.bucket = std::move(args.bucket);
    amu_args.region = std::move(args.region);
    amu_args.object = std::move(args.object);
    amu_args.upload_id = upload_id;
    AbortMultipartUpload(amu_args);
  }

  return resp;
}

UploadObjectResponse Client::UploadObject(UploadObjectArgs args) {
  if (error::Error err = args.Validate()) {
    return UploadObjectResponse(err);
  }

  std::ifstream file;
  file.exceptions(std::ifstream::failbit | std::ifstream::badbit);
  try {
    file.open(args.filename);
  } catch (std::system_error& err) {
    return error::make<UploadObjectResponse>(
        "unable to open file " + args.filename + "; " + err.code().message());
  }

  PutObjectArgs po_args(file, args.object_size, 0);
  po_args.extra_headers = std::move(args.extra_headers);
  po_args.extra_query_params = std::move(args.extra_query_params);
  po_args.bucket = std::move(args.bucket);
  po_args.region = std::move(args.region);
  po_args.object = std::move(args.object);
  po_args.headers = std::move(args.headers);
  po_args.user_metadata = std::move(args.user_metadata);
  po_args.sse = std::move(args.sse);
  po_args.tags = std::move(args.tags);
  po_args.retention = std::move(args.retention);
  po_args.legal_hold = std::move(args.legal_hold);
  po_args.content_type = std::move(args.content_type);
  po_args.progressfunc = std::move(args.progressfunc);
  po_args.progress_userdata = std::move(args.progress_userdata);

  PutObjectResponse resp = PutObject(std::move(po_args));
  file.close();
  return UploadObjectResponse(resp);
}

RemoveObjectsResult Client::RemoveObjects(RemoveObjectsArgs args) {
  if (error::Error err = args.Validate()) {
    return RemoveObjectsResult(err);
  }
  return RemoveObjectsResult(this, std::move(args));
}

// ---- Async overloads ----

std::future<ComposeObjectResponse> Client::ComposeObjectAsync(
    ComposeObjectArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return ComposeObject(std::move(args));
                    });
}

std::future<CopyObjectResponse> Client::CopyObjectAsync(CopyObjectArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return CopyObject(std::move(args));
                    });
}

std::future<DownloadObjectResponse> Client::DownloadObjectAsync(
    DownloadObjectArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return DownloadObject(std::move(args));
                    });
}

std::future<GetObjectResponse> Client::GetObjectAsync(GetObjectArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return Client::GetObject(std::move(args));
                    });
}

std::future<ListObjectsResult> Client::ListObjectsAsync(ListObjectsArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return ListObjects(std::move(args));
                    });
}

std::future<PutObjectResponse> Client::PutObjectAsync(PutObjectArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return PutObject(std::move(args));
                    });
}

std::future<UploadObjectResponse> Client::UploadObjectAsync(
    UploadObjectArgs args) {
  return std::async(std::launch::async,
                    [this, args = std::move(args)]() mutable {
                      return UploadObject(std::move(args));
                    });
}

}  // namespace minio::s3