RestCatalog: concurrent HTTP requests can crash in libcurl

[smaheshwar-pltr]

HttpClient sends every request through one cpr::ConnectionPool (http_client.cc#L158), which wraps a libcurl share handle with CURL_LOCK_DATA_CONNECT. libcurl does not support sharing connections between threads, so when two requests overlap, the process can crash inside libcurl.

A RestCatalog sends its requests through one HttpClient, so they can overlap even if the application only uses the catalog from one thread:

• when a metrics executor is passed to RestCatalog::Make, metrics reports are posted from its threads (rest_metrics_reporter.cc#L74);
• with OAuth2 client credentials, tokens are refreshed from TokenRefreshScheduler's thread (auth_session_internal.h#L205).

Applications that share a catalog across threads hit it directly (see #21). We found it in a DuckDB extension's concurrency test.

The shared pool came in with apache#530, which replaced a mutex-guarded cpr::Session. #19 gives each in-flight request its own pool. Reproduction in the first comment.

[smaheshwar-pltr]

Reproduction: add this to rest_catalog_integration_test.cc, build with -DICEBERG_BUILD_REST_INTEGRATION_TESTS=ON, and run rest_catalog_integration_test --gtest_filter='*MetricsExecutorCrash*'. The catalog is used from one thread, and metrics are posted from a 4-thread pool.

Test

Includes:

#include <condition_variable>
#include <deque>
#include <mutex>

#include "iceberg/expression/expressions.h"
#include "iceberg/metrics/scan_report.h"
#include "iceberg/util/executor.h"

Inside namespace iceberg::rest:

// Fixed-size thread pool.
class ThreadPool final : public Executor {
 public:
  explicit ThreadPool(int size) {
    for (int i = 0; i < size; ++i) workers_.emplace_back([this] { Work(); });
  }
  ~ThreadPool() override {
    {
      std::lock_guard lock(mutex_);
      stopping_ = true;
    }
    cv_.notify_all();
    for (auto& worker : workers_) worker.join();
  }
  Status Submit(ExecutorTask task) override {
    {
      std::lock_guard lock(mutex_);
      tasks_.push_back(std::move(task));
    }
    cv_.notify_one();
    return {};
  }

 private:
  void Work() {
    while (true) {
      std::unique_lock lock(mutex_);
      cv_.wait(lock, [this] { return stopping_ || !tasks_.empty(); });
      if (tasks_.empty()) return;
      ExecutorTask task = std::move(tasks_.front());
      tasks_.pop_front();
      lock.unlock();
      std::move(task)();
    }
  }

  std::mutex mutex_;
  std::condition_variable cv_;
  std::deque<ExecutorTask> tasks_;
  bool stopping_ = false;
  std::vector<std::thread> workers_;
};

TEST_F(RestCatalogIntegrationTest, MetricsExecutorCrash) {
  ThreadPool executor(4);
  auto config = RestCatalogProperties::default_properties();
  config.Set(RestCatalogProperties::kUri, CatalogUri())
      .Set(RestCatalogProperties::kName, std::string(kCatalogName))
      .Set(RestCatalogProperties::kWarehouse, std::string(kWarehouseName));
  config.mutable_configs()[std::string(RestCatalogProperties::kIOImpl.key())] =
      std::string(kStdFileIOImpl);
  ICEBERG_UNWRAP_OR_FAIL(auto root, RestCatalog::Make(config, &executor));
  ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog());

  Namespace ns{.levels = {"metrics_crash"}};
  ASSERT_THAT(catalog->CreateNamespace(ns, {}), IsOk());
  TableIdentifier id{.ns = ns, .name = "t"};
  ASSERT_THAT(CreateDefaultTable(catalog, id), IsOk());

  for (int i = 0; i < 2000; ++i) {
    ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(id));
    // The report a finished scan sends.
    ScanReport report{.table_name = "t", .snapshot_id = 1,
                      .filter = Expressions::AlwaysTrue(), .schema_id = 1};
    ASSERT_THAT(table->reporter()->Report(report), IsOk());
  }
}

Crashes out of 8 runs on 450863e, switching libcurl with LD_LIBRARY_PATH. The last column uses 32 threads sharing one catalog (no executor), each calling LoadTable 200 times.

libcurl	test above	32 threads calling LoadTable
8.22.0	8	4
8.18.0 (CI)	0	0
8.5.0	0	8

So CI doesn't catch this. Crash with 8.22.0:

#0  url_match_conn ()
#1  Curl_cpool_find ()
#2  Curl_connect ()
...
#11 iceberg::rest::HttpClient::Get(...)
#13 iceberg::rest::RestCatalog::LoadTable(...)