Copilot commented on code in PR #1033:
URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4303612304
##########
src/iceberg/catalog/rest/auth/auth_manager.cc:
##########
@@ -319,12 +323,14 @@ class OAuth2Manager : public AuthManager {
std::optional<OAuthTokenResponse> auth_response_;
std::optional<std::chrono::steady_clock::time_point> start_time_;
std::shared_ptr<HttpClient> refresh_client_;
+ std::unique_ptr<::iceberg::internal::ScheduledExecutor> scheduler_;
Review Comment:
Refreshing sessions retain raw pointers to this manager-owned scheduler in
their configuration and cancellation handles. Destroying the manager before a
returned session, or calling `SetAuthRefreshExecutor()` again, frees the
scheduler while those pointers remain live. The session's next refresh or
`Close()` can then access freed memory, even if the injected executor still
exists. Give sessions lifetime-safe scheduler ownership/state, or close and
detach all sessions before disposing of the scheduler.
##########
src/iceberg/catalog/rest/auth/auth_session_internal.h:
##########
@@ -118,10 +119,10 @@ class OAuth2Session final : public AuthSession,
if (!closed_.compare_exchange_strong(expected, true)) {
return {};
}
- TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0));
+ scheduled_task_.Cancel();
Review Comment:
Synchronize all accesses to `scheduled_task_`. This first cancellation runs
before waiting for active refreshes, so a refresh that already passed its
`closed_` check can concurrently assign the handle in `ScheduleRefresh()` or
the retry path. Unlike the previous atomic ID, the handle has ordinary
pointer/ID fields that `Cancel()` also modifies, causing undefined behavior.
Protect both cancellations and both assignment paths with a common mutex,
without holding it across the active-refresh wait.
##########
src/iceberg/util/scheduled_executor_internal.h:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <unordered_set>
+#include <vector>
+
+#include "iceberg/result.h"
+#include "iceberg/util/executor.h"
+
+namespace iceberg::internal {
+
+class ScheduledExecutor;
+
+class ScheduledTask {
+ public:
+ ScheduledTask() = default;
+ ScheduledTask(ScheduledExecutor* executor, uint64_t id)
+ : executor_(executor), id_(id) {}
+
+ void Cancel();
+ bool valid() const { return executor_ != nullptr && id_ != 0; }
+
+ private:
+ ScheduledExecutor* executor_ = nullptr;
+ uint64_t id_ = 0;
+};
+
+/// A delay queue that dispatches due tasks to an existing Executor.
+///
+/// The executor is borrowed and must outlive this object and all scheduled
tasks.
+class ScheduledExecutor {
Review Comment:
Export both `ScheduledExecutor` and `ScheduledTask` with `ICEBERG_EXPORT`.
Their out-of-line methods are compiled into the core library but called from
`iceberg_rest` and its tests. Shared libraries use hidden visibility
(`cmake_modules/IcebergBuildUtils.cmake:180–183`), so these symbols are
unavailable across that boundary and shared-library consumers fail to link.
##########
src/iceberg/util/scheduled_executor.cc:
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include <optional>
+#include <utility>
+
+#include "iceberg/util/scheduled_executor_internal.h"
+
+namespace iceberg::internal {
+
+void ScheduledTask::Cancel() {
+ if (valid()) {
+ executor_->Cancel(id_);
+ executor_ = nullptr;
+ id_ = 0;
+ }
+}
+
+ScheduledExecutor::ScheduledExecutor(Executor& executor)
+ : executor_(executor), timer_thread_([this] { Run(); }) {}
+
+ScheduledExecutor::~ScheduledExecutor() { Shutdown(); }
+
+ScheduledTask
ScheduledExecutor::Schedule(std::chrono::steady_clock::time_point deadline,
+ ExecutorTask task) {
+ std::lock_guard lock(mutex_);
+ if (shutdown_) return {};
+ auto id = next_id_++;
+ tasks_.push(Task{.id = id, .deadline = deadline, .task = std::move(task)});
+ scheduled_.insert(id);
+ cv_.notify_one();
+ return ScheduledTask(this, id);
+}
+
+ScheduledTask ScheduledExecutor::Schedule(std::chrono::milliseconds delay,
+ ExecutorTask task) {
+ return Schedule(std::chrono::steady_clock::now() + delay, std::move(task));
+}
+
+void ScheduledExecutor::Cancel(uint64_t id) {
+ if (id == 0) return;
+ {
+ std::lock_guard lock(mutex_);
+ if (scheduled_.erase(id) != 0) cancelled_.insert(id);
+ }
+ cv_.notify_one();
+}
+
+void ScheduledExecutor::Shutdown() {
+ {
+ std::lock_guard lock(mutex_);
+ if (shutdown_) return;
+ shutdown_ = true;
+ while (!tasks_.empty()) tasks_.pop();
+ cancelled_.clear();
+ scheduled_.clear();
+ }
+ cv_.notify_one();
+ if (timer_thread_.joinable()) timer_thread_.join();
+}
+
+void ScheduledExecutor::Run() {
+ while (true) {
+ std::optional<Task> task;
+ {
+ std::unique_lock lock(mutex_);
+ cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); });
+ if (shutdown_) return;
+
+ while (!tasks_.empty() && cancelled_.contains(tasks_.top().id)) {
+ cancelled_.erase(tasks_.top().id);
+ tasks_.pop();
+ }
+ if (tasks_.empty()) continue;
+
+ auto deadline = tasks_.top().deadline;
+ if (cv_.wait_until(lock, deadline, [this, deadline] {
+ return shutdown_ || tasks_.empty() || tasks_.top().deadline !=
deadline ||
+ cancelled_.contains(tasks_.top().id);
+ })) {
+ continue;
+ }
+
+ task.emplace(std::move(const_cast<Task&>(tasks_.top())));
+ tasks_.pop();
+ scheduled_.erase(task->id);
+ if (cancelled_.erase(task->id) != 0) continue;
+ }
+ std::ignore = executor_.Submit(std::move(task->task));
Review Comment:
Handle submission failures instead of discarding the status. If the injected
executor rejects a refresh task, that task has already been removed from the
scheduler queue. The callback never starts, so neither OAuth retry logic nor
the next refresh runs, silently stopping refresh until the cached token
expires. Preserve a way to retry dispatch or notify the owning session of
rejection.
##########
src/iceberg/catalog/rest/auth/auth_manager.h:
##########
@@ -38,6 +38,10 @@ class ICEBERG_REST_EXPORT AuthManager {
public:
virtual ~AuthManager() = default;
+ /// Configure the borrowed executor used for background authentication
refresh.
+ /// The executor must outlive this manager and all sessions it creates.
+ virtual void SetAuthRefreshExecutor(Executor* /*executor*/) {}
Review Comment:
Forward this hook through `SigV4AuthManager`. That wrapper defaults to an
OAuth2 delegate but inherits this no-op, so `RestCatalog::Make(...,
auth_executor)` never configures the delegate's refresh executor. Combined
SigV4/OAuth2 catalogs therefore stop refreshing expiring tokens even when an
executor is supplied. Add a `SigV4AuthManager` override that calls
`delegate_->SetAuthRefreshExecutor(executor)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
