MINIFICPP-2705 - Configurable timeout, download assets directly to disk

adamdebreceni · szaszm · commit af585c264551 · 2026-01-29T16:17:14.000+01:00
Closes #2088 Signed-off-by: Marton Szasz <szaszm@apache.org>
diff --git a/C2.md b/C2.md
@@ -115,6 +115,12 @@ be requested via C2 DESCRIBE manifest command.
     # specify the maximum number of bulletins to send in a heartbeat
     # nifi.c2.flow.info.processor.bulletin.limit=1000
 
+    # Specify timeout for asset download operations. The entire download must
+    # finish in the specified amount of time. There is a separate fixed 30 second
+    # timeout from the last received data packet.
+    # setting to 0 disables the timeout (default)
+    nifi.c2.asset.download.timeout=0s
+
 #### Flow Id and URL
 
 Flow id and URL are usually retrieved from the C2 server. These identify the last updated flow version and where the flow was downloaded from. These properties are persisted in the minifi.properties file.
diff --git a/conf/minifi.properties.in b/conf/minifi.properties.in
@@ -127,6 +127,12 @@ nifi.c2.full.heartbeat=false
 # specify the maximum number of bulletins to send in a heartbeat
 #nifi.c2.flow.info.processor.bulletin.limit=1000
 
+# Specify timeout for asset download operations. The entire download must
+# finish in the specified amount of time. There is a separate fixed 30 second
+# timeout from the last received data packet.
+# setting to 0 disables the timeout (default)
+# nifi.c2.asset.download.timeout=0s
+
 ## enable the controller socket provider on port 9998
 ## off by default.
 #controller.socket.enable=true
diff --git a/core-framework/include/http/BaseHTTPClient.h b/core-framework/include/http/BaseHTTPClient.h
@@ -106,11 +106,23 @@ class HTTPUploadStreamContentsCallback : public HTTPUploadCallback {
   std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<HTTPUploadStreamContentsCallback>::getLogger();
 };
 
-class HTTPReadCallback : public utils::ByteOutputCallback {
+class HTTPReadCallback {
  public:
-  using ByteOutputCallback::ByteOutputCallback;
+  virtual bool process(std::span<const char> data) = 0;
+  virtual ~HTTPReadCallback() = default;
 
   std::atomic<bool> stop = false;
+};
+
+class HTTPReadByteOutputCallback : public HTTPReadCallback, public utils::ByteOutputCallback {
+ public:
+  using ByteOutputCallback::ByteOutputCallback;
+
+  bool process(std::span<const char> data) override {
+    ByteOutputCallback::write(data.data(), data.size());
+    return true;
+  }
+
   std::atomic<size_t> pos = 0;
 };
 
diff --git a/core-framework/include/http/HTTPClient.h b/core-framework/include/http/HTTPClient.h
@@ -99,6 +99,10 @@ class HTTPClient : public BaseHTTPClient, public core::ConnectableImpl {
 
   void setReadTimeout(std::chrono::milliseconds timeout) override;
 
+  void setAbsoluteTimeout(std::optional<std::chrono::milliseconds> timeout) {
+    absolute_timeout_ = timeout;
+  }
+
   void setUploadCallback(std::unique_ptr<HTTPUploadCallback> callback) override;
 
   void setReadCallback(std::unique_ptr<HTTPReadCallback> callback);
@@ -226,16 +230,15 @@ class HTTPClient : public BaseHTTPClient, public core::ConnectableImpl {
 
   void configure_secure_connection();
 
-  std::chrono::milliseconds getAbsoluteTimeout() const { return 3*read_timeout_; }
-
-  HTTPReadCallback content_{std::numeric_limits<size_t>::max()};
+  HTTPReadByteOutputCallback content_{std::numeric_limits<size_t>::max()};
 
   std::shared_ptr<minifi::controllers::SSLContextServiceInterface> ssl_context_service_;
   std::string url_;
   std::optional<http::HttpRequestMethod> method_;
 
   std::chrono::milliseconds connect_timeout_{std::chrono::seconds(30)};
   std::chrono::milliseconds read_timeout_{std::chrono::seconds(30)};
+  std::optional<std::chrono::milliseconds> absolute_timeout_;
 
   HTTPResponseData response_data_;
 
diff --git a/core-framework/include/http/HTTPStream.h b/core-framework/include/http/HTTPStream.h
@@ -115,9 +115,9 @@ class HttpStream : public io::BaseStreamImpl {
 
   inline bool isFinished(int seconds = 0) {
     return http_client_future_.wait_for(std::chrono::seconds(seconds)) == std::future_status::ready
-        && http_client_->getReadCallback()
-        && http_client_->getReadCallback()->getSize() == 0
-        && http_client_->getReadCallback()->waitingOps();
+        && getByteOutputReadCallback()
+        && getByteOutputReadCallback()->getSize() == 0
+        && getByteOutputReadCallback()->waitingOps();
   }
 
   /**
@@ -127,11 +127,11 @@ class HttpStream : public io::BaseStreamImpl {
     do {
       logger_->log_trace("Waiting for more data");
     } while (http_client_future_.wait_for(std::chrono::seconds(0)) != std::future_status::ready
-        && http_client_->getReadCallback()
-        && http_client_->getReadCallback()->getSize() == 0);
+        && getByteOutputReadCallback()
+        && getByteOutputReadCallback()->getSize() == 0);
 
-    return http_client_->getReadCallback()
-        && http_client_->getReadCallback()->getSize() > 0;
+    return getByteOutputReadCallback()
+        && getByteOutputReadCallback()->getSize() > 0;
   }
 
  protected:
@@ -147,6 +147,10 @@ class HttpStream : public io::BaseStreamImpl {
   std::atomic<bool> started_{false};
 
  private:
+  utils::ByteOutputCallback* getByteOutputReadCallback() {
+    return dynamic_cast<utils::ByteOutputCallback*>(http_client_->getReadCallback());
+  }
+
   std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<HttpStream>::getLogger();
 };
 }  // namespace org::apache::nifi::minifi::http
diff --git a/core-framework/include/utils/ByteArrayCallback.h b/core-framework/include/utils/ByteArrayCallback.h
@@ -111,12 +111,12 @@ class ByteOutputCallback {
 
   bool waitingOps();
 
-  virtual void write(char *data, size_t size);
+  virtual void write(const char *data, size_t size);
 
   size_t readFully(char *buffer, size_t size);
 
  protected:
-  inline void write_and_notify(char *data, size_t size);
+  inline void write_and_notify(const char *data, size_t size);
 
   inline size_t read_current_str(char *buffer, size_t size);
 
diff --git a/core-framework/src/http/BaseHTTPClient.cpp b/core-framework/src/http/BaseHTTPClient.cpp
@@ -173,8 +173,10 @@ size_t HTTPRequestResponse::receiveWrite(char *data, size_t size, size_t nmemb,
     if (callback->stop) {
       return CALLBACK_ABORT;
     }
-    callback->write(data, (size * nmemb));
-    return (size * nmemb);
+    if (!callback->process(std::span(data, size * nmemb))) {
+      return CALLBACK_ABORT;
+    }
+    return size * nmemb;
   } catch (...) {
     return CALLBACK_ABORT;
   }
diff --git a/core-framework/src/http/HTTPClient.cpp b/core-framework/src/http/HTTPClient.cpp
@@ -334,9 +334,11 @@ bool HTTPClient::submit() {
 
   response_data_.clear();
 
+  const auto absolute_timeout = absolute_timeout_.value_or(3 * read_timeout_);
+
   curl_easy_setopt(http_session_.get(), CURLOPT_NOSIGNAL, 1);
   curl_easy_setopt(http_session_.get(), CURLOPT_CONNECTTIMEOUT_MS, connect_timeout_.count());
-  curl_easy_setopt(http_session_.get(), CURLOPT_TIMEOUT_MS, getAbsoluteTimeout().count());
+  curl_easy_setopt(http_session_.get(), CURLOPT_TIMEOUT_MS, absolute_timeout.count());
 
   if (read_timeout_ > 0ms) {
     progress_.reset();
@@ -376,7 +378,7 @@ bool HTTPClient::submit() {
   response_data_.response_code = http_code;
   curl_easy_getinfo(http_session_.get(), CURLINFO_CONTENT_TYPE, &response_data_.response_content_type);
   if (res_ == CURLE_OPERATION_TIMEDOUT) {
-    logger_->log_error("HTTP operation timed out, with absolute timeout {}\n", getAbsoluteTimeout());
+    logger_->log_error("HTTP operation timed out, with absolute timeout {}\n", absolute_timeout);
   }
   if (res_ != CURLE_OK) {
     logger_->log_info("{}", request_headers_.size());
@@ -398,8 +400,8 @@ const char *HTTPClient::getContentType() {
 
 const std::vector<char> &HTTPClient::getResponseBody() {
   if (response_data_.response_body.empty()) {
-    if (read_callback_) {
-      response_data_.response_body = read_callback_->to_string();
+    if (auto byte_output_callback = dynamic_cast<utils::ByteOutputCallback*>(read_callback_.get())) {
+      response_data_.response_body = byte_output_callback->to_string();
     } else {
       response_data_.response_body = content_.to_string();
     }
diff --git a/core-framework/src/http/HTTPStream.cpp b/core-framework/src/http/HTTPStream.cpp
@@ -33,7 +33,7 @@ HttpStream::HttpStream(std::shared_ptr<HTTPClient> client)
 }
 
 void HttpStream::close() {
-  if (auto read_callback = http_client_->getReadCallback())
+  if (auto read_callback = getByteOutputReadCallback())
     read_callback->close();
   if (auto upload_callback = http_client_->getUploadCallback())
     upload_callback->close();
@@ -78,13 +78,13 @@ size_t HttpStream::read(std::span<std::byte> buf) {
     if (!started_) {
       std::lock_guard<std::mutex> lock(mutex_);
       if (!started_) {
-        auto read_callback = std::make_unique<HTTPReadCallback>(66560, true);
+        auto read_callback = std::make_unique<HTTPReadByteOutputCallback>(66560, true);
         http_client_future_ = std::async(std::launch::async, submit_read_client, http_client_, read_callback.get());
         http_client_->setReadCallback(std::move(read_callback));
         started_ = true;
       }
     }
-    return http_client_->getReadCallback()->readFully(reinterpret_cast<char*>(buf.data()), buf.size());
+    return gsl::not_null(getByteOutputReadCallback())->readFully(reinterpret_cast<char*>(buf.data()), buf.size());
   } else {
     return io::STREAM_ERROR;
   }
diff --git a/core-framework/src/utils/ByteArrayCallback.cpp b/core-framework/src/utils/ByteArrayCallback.cpp
@@ -62,7 +62,7 @@ bool ByteOutputCallback::waitingOps() {
   return true;
 }
 
-void ByteOutputCallback::write(char *data, size_t size) {
+void ByteOutputCallback::write(const char *data, size_t size) {
   if (!read_started_) {
     std::unique_lock<std::recursive_mutex> lock(vector_lock_);
     spinner_.wait(lock, [&] {
@@ -73,7 +73,7 @@ void ByteOutputCallback::write(char *data, size_t size) {
   write_and_notify(data, size);
 }
 
-void ByteOutputCallback::write_and_notify(char *data, size_t size) {
+void ByteOutputCallback::write_and_notify(const char *data, size_t size) {
   queue_.enqueue(std::string(data, size));
   size_ += size;
   total_written_ += size;
diff --git a/libminifi/include/c2/C2Protocol.h b/libminifi/include/c2/C2Protocol.h
@@ -72,8 +72,10 @@ class C2Protocol : public core::ConnectableImpl {
    */
   virtual C2Payload consumePayload(const C2Payload &operation, Direction direction = TRANSMIT, bool async = false) = 0;
 
-  virtual C2Payload fetch(const std::string& url, const std::vector<std::string>& /*accepted_formats*/ = {}, bool async = false) {
-    return consumePayload(url, C2Payload(Operation::transfer, true), Direction::RECEIVE, async);
+  virtual bool fetch(const std::string& url, const std::vector<std::string>& accepted_formats, std::function<bool(std::span<const char> chunk)> chunk_callback) = 0;
+
+  bool fetch(const std::string& url, std::function<bool(std::span<const char> chunk)> chunk_callback) {
+    return fetch(url, {}, std::move(chunk_callback));
   }
 
   /**
diff --git a/libminifi/include/c2/protocols/RESTSender.h b/libminifi/include/c2/protocols/RESTSender.h
@@ -52,7 +52,7 @@ class RESTSender : public RESTProtocol, public C2Protocol {
 
   C2Payload consumePayload(const C2Payload &payload, Direction direction, bool async) override;
 
-  C2Payload fetch(const std::string& url, const std::vector<std::string>& accepted_formats, bool async) override;
+  bool fetch(const std::string& url, const std::vector<std::string>& accepted_formats, std::function<bool(std::span<const char> chunk)> chunk_callback) override;
 
   void update(const std::shared_ptr<Configure> &configure) override;
 
@@ -77,6 +77,7 @@ class RESTSender : public RESTProtocol, public C2Protocol {
   RequestEncoding req_encoding_ = RequestEncoding::none;
 
  private:
+  std::chrono::milliseconds asset_download_timeout_{0};
   std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<RESTSender>::getLogger();
 };
 
diff --git a/libminifi/include/utils/file/AssetManager.h b/libminifi/include/utils/file/AssetManager.h
@@ -53,7 +53,8 @@ class AssetManager {
  public:
   explicit AssetManager(const Configure& configuration);
 
-  nonstd::expected<void, std::string> sync(const AssetLayout& layout, const std::function<nonstd::expected<std::vector<std::byte>, std::string>(std::string_view /*url*/)>& fetch);
+  nonstd::expected<void, std::string> sync(const AssetLayout& layout,
+      const std::function<nonstd::expected<void, std::string>(std::string_view /*url*/, const std::filesystem::path& /*tmp_path*/)>& fetch);
 
   std::string hash() const;
 
diff --git a/libminifi/src/c2/C2Agent.cpp b/libminifi/src/c2/C2Agent.cpp
@@ -321,6 +321,33 @@ struct C2DebugBundleError : public C2TransferError {
   using C2TransferError::C2TransferError;
 };
 
+nonstd::expected<void, std::string> fetchAssetAsFile(C2Protocol& protocol, const std::string& url, const std::filesystem::path& file_path, const std::shared_ptr<core::logging::Logger>& logger) {
+  if (utils::file::create_dir(file_path.parent_path()) != 0) {
+    return nonstd::make_unexpected(fmt::format("Failed to create directory '{}'", file_path.parent_path().string()));
+  }
+
+  std::ofstream file{file_path, std::ofstream::binary};
+  if (!file) {
+    return nonstd::make_unexpected(fmt::format("Failed to open file to write '{}'", file_path.string()));
+  }
+  bool success = protocol.fetch(url, [&] (std::span<const char> chunk) {
+    file.write(chunk.data(), gsl::narrow<std::streamsize>(chunk.size()));
+    return file.good();
+  });
+  file.close();
+  if (!file || !success) {
+    std::error_code ec;
+    std::filesystem::remove(file_path, ec);
+    if (ec) {
+      logger->log_error("Failed remove partial asset file '{}'", file_path.string());
+    } else {
+      logger->log_info("Successfully removed partial asset file '{}'", file_path.string());
+    }
+    return nonstd::make_unexpected(fmt::format("Failed to fetch asset from '{}'", url));
+  }
+  return {};
+}
+
 }  // namespace
 
 void C2Agent::handle_c2_server_response(const C2ContentResponse &resp) {
@@ -828,21 +855,13 @@ void C2Agent::handle_sync(const org::apache::nifi::minifi::c2::C2ContentResponse
     });
   }
 
-  auto fetch = [&] (std::string_view url) -> nonstd::expected<std::vector<std::byte>, std::string> {
+  auto result = asset_manager_->sync(asset_layout, [&] (std::string_view url, const std::filesystem::path& file_path) -> nonstd::expected<void, std::string> {
     auto resolved_url = resolveUrl(std::string{url});
     if (!resolved_url) {
       return nonstd::make_unexpected("Couldn't resolve url");
     }
-    C2Payload file_response = protocol_->fetch(resolved_url.value());
-
-    if (file_response.getStatus().getState() != state::UpdateState::READ_COMPLETE) {
-      return nonstd::make_unexpected("Failed to fetch file from " + resolved_url.value());
-    }
-
-    return std::move(file_response).moveRawData();
-  };
-
-  auto result = asset_manager_->sync(asset_layout, fetch);
+    return fetchAssetAsFile(*protocol_, resolved_url.value(), file_path, logger_);
+  });
   if (!result) {
     send_error(result.error());
     return;
@@ -1012,9 +1031,18 @@ std::optional<std::string> C2Agent::fetchFlow(const std::string& uri) const {
     return std::nullopt;
   }
 
-  C2Payload response = protocol_->fetch(resolved_url.value(), update_sink_->getSupportedConfigurationFormats());
+  std::string flow_content;
+  bool success = protocol_->fetch(resolved_url.value(), update_sink_->getSupportedConfigurationFormats(), [&] (std::span<const char> chunk) {
+    flow_content.append(chunk.data(), chunk.size());
+    return true;
+  });
+
+  if (!success) {
+    logger_->log_error("Could not fetch flow content from '{}'", uri);
+    return std::nullopt;
+  }
 
-  return response.getRawDataAsString();
+  return flow_content;
 }
 
 std::optional<std::string> C2Agent::getFlowIdFromConfigUpdate(const C2ContentResponse &resp) {
@@ -1140,25 +1168,21 @@ void C2Agent::handleAssetUpdate(const C2ContentResponse& resp) {
     return;
   }
 
-  C2Payload file_response = protocol_->fetch(url);
+  std::filesystem::path tmp_file{file_path.string() + ".part"};
 
-  if (file_response.getStatus().getState() != state::UpdateState::READ_COMPLETE) {
-    send_error("Failed to fetch asset from '" + url + "'");
+  auto fetch_status = fetchAssetAsFile(*protocol_, url, tmp_file, logger_);
+  if (!fetch_status) {
+    send_error(fetch_status.error());
     return;
   }
 
-  auto raw_data = std::move(file_response).moveRawData();
-  // ensure directory exists for file
-  if (utils::file::create_dir(file_path.parent_path()) != 0) {
-    send_error("Failed to create directory '" + file_path.parent_path().string() + "'");
+  std::error_code ec;
+  std::filesystem::rename(tmp_file, file_path, ec);
+  if (ec) {
+    send_error(fmt::format("Failed to move temporary asset file '{}' to '{}'", tmp_file.string(), file_path.string()));
     return;
   }
 
-  {
-    std::ofstream file{file_path, std::ofstream::binary};
-    file.write(reinterpret_cast<const char*>(raw_data.data()), gsl::narrow<std::streamsize>(raw_data.size()));
-  }
-
   C2Payload response(Operation::acknowledge, state::UpdateState::FULLY_APPLIED, resp.ident, true);
   enqueue_c2_response(std::move(response));
 }
diff --git a/libminifi/src/c2/protocols/RESTSender.cpp b/libminifi/src/c2/protocols/RESTSender.cpp
diff --git a/libminifi/src/utils/file/AssetManager.cpp b/libminifi/src/utils/file/AssetManager.cpp
diff --git a/minifi-api/include/minifi-cpp/properties/Configuration.h b/minifi-api/include/minifi-cpp/properties/Configuration.h

Original file line number	Diff line number	Diff line change
`@@ -173,8 +173,10 @@ size_t HTTPRequestResponse::receiveWrite(char *data, size_t size, size_t nmemb,`
`173`	`173`	`if (callback->stop) {`
`174`	`174`	`return CALLBACK_ABORT;`
`175`	`175`	`}`
`176`		`- callback->write(data, (size * nmemb));`
`177`		`- return (size * nmemb);`
	`176`	`+ if (!callback->process(std::span(data, size * nmemb))) {`
	`177`	`+ return CALLBACK_ABORT;`
	`178`	`+ }`
	`179`	`+ return size * nmemb;`
`178`	`180`	`} catch (...) {`
`179`	`181`	`return CALLBACK_ABORT;`
`180`	`182`	`}`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ HttpStream::HttpStream(std::shared_ptr<HTTPClient> client)`
`33`	`33`	`}`
`34`	`34`
`35`	`35`	`void HttpStream::close() {`
`36`		`- if (auto read_callback = http_client_->getReadCallback())`
	`36`	`+ if (auto read_callback = getByteOutputReadCallback())`
`37`	`37`	`read_callback->close();`
`38`	`38`	`if (auto upload_callback = http_client_->getUploadCallback())`
`39`	`39`	`upload_callback->close();`
`@@ -78,13 +78,13 @@ size_t HttpStream::read(std::span<std::byte> buf) {`
`78`	`78`	`if (!started_) {`
`79`	`79`	`std::lock_guard<std::mutex> lock(mutex_);`
`80`	`80`	`if (!started_) {`
`81`		`- auto read_callback = std::make_unique<HTTPReadCallback>(66560, true);`
	`81`	`+ auto read_callback = std::make_unique<HTTPReadByteOutputCallback>(66560, true);`
`82`	`82`	`http_client_future_ = std::async(std::launch::async, submit_read_client, http_client_, read_callback.get());`
`83`	`83`	`http_client_->setReadCallback(std::move(read_callback));`
`84`	`84`	`started_ = true;`
`85`	`85`	`}`
`86`	`86`	`}`
`87`		`- return http_client_->getReadCallback()->readFully(reinterpret_cast<char*>(buf.data()), buf.size());`
	`87`	`+ return gsl::not_null(getByteOutputReadCallback())->readFully(reinterpret_cast<char*>(buf.data()), buf.size());`
`88`	`88`	`} else {`
`89`	`89`	`return io::STREAM_ERROR;`
`90`	`90`	`}`