http: worker errors from an HTTP error body carry the call's context

Over HTTP a worker error raised during a request comes back in the response
body (X-VGI-RPC-Error), decoded by the shared POST helper with nothing but
the request URL and no ClientContext. So, unlike the same error on a
subprocess stream, it carried no execution_id, function_name or
function_kind, reported the request route (".../vgi.v2/init/exchange") as
worker_path, and dropped the worker's log lines that preceded the failure
instead of sending them to duckdb_logs.

Callers now pass an RpcErrorContext (worker base URL, execution, function
fields) through HttpPostArrowIpc / HttpInvokeUnary to the decoder; the
request route is reported as `url` and the method as `rpc_method`. An init
reply that fails right after init carries the newly minted execution_id in
its header stream, which the decoder adopts as it walks past. Requests
without a context (the cancel dispatcher's, off the query thread) behave as
before.

error_info.test asserts execution_id and function_name on a mid-scan worker
error, on every transport and SDK; it fails over HTTP without this change.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: rustyconover

src/include/vgi_http_client.hpp

@@ -69,7 +69,8 @@ UnaryResponseResult HttpInvokeUnary(ClientContext &context,
                                      const VgiProtocolId &protocol = VGI_MAIN_PROTOCOL,
                                      duckdb::unique_ptr<HTTPClient> *client_holder = nullptr,
                                      ServerCapabilities *caps = nullptr,
-                                     const std::shared_ptr<IrohClientConfig> &iroh_config = nullptr);
+                                     const std::shared_ptr<IrohClientConfig> &iroh_config = nullptr,
+                                     const std::unordered_map<std::string, std::string> *error_fields = nullptr);
 
 // Small thread-safe pool of keep-alive HTTPClients for unary RPCs against ONE
 // base URL (a pool instance lives on a catalog's VgiAttachParameters, whose
@@ -152,7 +153,8 @@ std::string HttpPostArrowIpc(ClientContext &context,
                               duckdb::unique_ptr<HTTPClient> *client_holder = nullptr,
                               ServerCapabilities *harvested_caps = nullptr,
                               const std::shared_ptr<IrohClientConfig> &iroh_config = nullptr,
-                              const std::atomic<bool> *cancellation = nullptr);
+                              const std::atomic<bool> *cancellation = nullptr,
+                              const RpcErrorContext *error_context = nullptr);
 
 // HTTP GET raw bytes from a URL. Used for fetching externalized batches.
 // Handles X-VGI-Content-Encoding decompression (zstd or gzip). No auth

src/include/vgi_rpc_client.hpp

@@ -8,6 +8,7 @@
 #include <memory>
 #include <functional>
 #include <string>
+#include <unordered_map>
 #include <string_view>
 #include <utility>
 #include <vector>
@@ -341,6 +342,19 @@ UnaryResponseResult ReadUnaryResponseFromBuffer(const uint8_t *data, size_t len,
                                                  const std::string &transaction_opaque_data_digest = "",
                                                  const std::string &conn_id_hex = "");
 
+// What a caller knows about the call an error belongs to, which the decoder of
+// a worker error body does not: the worker it is (as opposed to the request
+// URL), the function execution, and fields to attach to the error (function
+// name / kind). Carried to HandleBatchLogMessage so a worker error decoded
+// from an HTTP error body is reported like one read off a subprocess stream.
+struct RpcErrorContext {
+	std::string worker_path; // the worker's base location; empty = the request URL
+	std::string execution_id_hex;
+	std::string attach_opaque_data_digest;
+	std::string conn_id_hex;
+	std::unordered_map<std::string, std::string> fields;
+};
+
 // Dispatch every Arrow IPC stream concatenated in an error response body,
 // throwing on the first batch that carries error metadata.
 //
@@ -357,7 +371,7 @@ UnaryResponseResult ReadUnaryResponseFromBuffer(const uint8_t *data, size_t len,
 // instead of a generic transport failure. Safe for a unary body too, which is
 // simply one stream.
 void DispatchErrorStreamsFromBuffer(const uint8_t *data, size_t len, ClientContext *context,
-                                    const std::string &url = "");
+                                    const std::string &url = "", const RpcErrorContext *call = nullptr);
 
 // Move-in overload: adopts the response body string as the owning Arrow buffer
 // (arrow::Buffer::FromString) instead of allocating + memcpying a copy of the

src/vgi_http_client.cpp

@@ -312,7 +312,8 @@ static std::string HttpPostArrowIpcInternal(
     bool allow_codec_retry = true, duckdb::unique_ptr<HTTPClient> *client_holder = nullptr,
                                              ServerCapabilities *harvested_caps = nullptr,
                                              const std::shared_ptr<IrohClientConfig> &iroh_config = nullptr,
-                                             const std::atomic<bool> *cancellation = nullptr) {
+                                             const std::atomic<bool> *cancellation = nullptr,
+                                             const RpcErrorContext *error_context = nullptr) {
 	const bool is_httpi = IsHttpiTransport(url);
 	const uint64_t timeout_seconds = GetHttpTimeoutSeconds(context);
 	const int64_t accepted_max_response_bytes = GetAcceptedMaxResponseBytes(context);
@@ -539,7 +540,7 @@ static std::string HttpPostArrowIpcInternal(
 			return HttpPostArrowIpcInternal(context, url, body, bearer_token, cookie_jar, out_response,
 			                                cached_http_params, alternate,
 			                                /*allow_codec_retry=*/false, client_holder, harvested_caps, iroh_config,
-			                                cancellation);
+			                                cancellation, error_context);
 		}
 	}
 
@@ -740,8 +741,14 @@ static std::string HttpPostArrowIpcInternal(
 			// header, returns without throwing, and loses the message to the
 			// generic throw below. DispatchErrorStreamsFromBuffer dispatches
 			// the error batch, which throws with the worker's own message.
+			//
+			// With a caller-supplied error context the errors name the worker,
+			// the execution and the function, and the worker's own log lines
+			// before the failure reach duckdb_logs, as on a subprocess stream.
+			// Without one (e.g. the cancel dispatcher's request, which runs off
+			// the query thread) nothing is logged, as before.
 			DispatchErrorStreamsFromBuffer(reinterpret_cast<const uint8_t *>(error_body.data()), error_body.size(),
-			                               nullptr, url);
+			                               error_context ? &context : nullptr, url, error_context);
 		}
 		throw IOException(BuildHttpExtraInfo(http_error::kProtocolViolation, url),
 		                  "VGI HTTP RPC error [url: %s]", url);
@@ -756,7 +763,7 @@ std::string HttpPostArrowIpc(ClientContext &context, const std::string &url, con
                               const std::shared_ptr<HTTPParams> &cached_http_params,
                              duckdb::unique_ptr<HTTPClient> *client_holder, ServerCapabilities *harvested_caps,
                              const std::shared_ptr<IrohClientConfig> &iroh_config,
-                             const std::atomic<bool> *cancellation) {
+                             const std::atomic<bool> *cancellation, const RpcErrorContext *error_context) {
 	// Get cached token from per-catalog auth (if any)
 	std::string token;
 	if (auth) {
@@ -772,7 +779,7 @@ std::string HttpPostArrowIpc(ClientContext &context, const std::string &url, con
 	std::unique_ptr<RpcHttpResponse> response;
 	auto result = HttpPostArrowIpcInternal(context, url, body, token, cookie_jar, response, cached_http_params,
 	                                       request_encoding, /*allow_codec_retry=*/true, client_holder, harvested_caps,
-	                                       iroh_config, cancellation);
+	                                       iroh_config, cancellation, error_context);
 
 	if (response->Status() != HTTPStatusCode::Unauthorized_401) {
 		return result;
@@ -845,7 +852,7 @@ std::string HttpPostArrowIpc(ClientContext &context, const std::string &url, con
 	result = HttpPostArrowIpcInternal(context, url, body, new_token, cookie_jar, response, cached_http_params,
 	                                  harvested_caps ? ChooseRequestEncoding(*harvested_caps) : request_encoding,
 	                                  /*allow_codec_retry=*/true, client_holder, harvested_caps, iroh_config,
-	                                  cancellation);
+	                                  cancellation, error_context);
 	if (response->Status() == HTTPStatusCode::Unauthorized_401) {
 		throw IOException(BuildHttpExtraInfo(http_error::kAuthFailed, url, 401),
 		                  "VGI HTTP authentication failed after auth flow (HTTP 401) [url: %s]. "
@@ -865,7 +872,8 @@ UnaryResponseResult HttpInvokeUnary(ClientContext &context, const std::string &w
                                     const std::string &transaction_opaque_data_digest, const std::string &conn_id_hex,
                                      const VgiProtocolId &protocol,
                                     duckdb::unique_ptr<HTTPClient> *client_holder, ServerCapabilities *caps,
-                                    const std::shared_ptr<IrohClientConfig> &iroh_config) {
+                                    const std::shared_ptr<IrohClientConfig> &iroh_config,
+                                    const std::unordered_map<std::string, std::string> *error_fields) {
 	std::string base_url = NormalizeBaseUrl(worker_path);
 	// Route shape: {base}/{protocol}/{method}. The protocol segment is the
 	// projection of the vgi_rpc.protocol routing key the body also carries --
@@ -905,9 +913,21 @@ UnaryResponseResult HttpInvokeUnary(ClientContext &context, const std::string &w
 	}
 
 	// POST to the route built above, using the standard HTTP timeout
+	// A worker error answered in the body names the worker and this call, not
+	// just the request URL; the URL rides along as its own field.
+	RpcErrorContext error_context;
+	error_context.worker_path = base_url;
+	error_context.execution_id_hex = execution_id_hex;
+	error_context.attach_opaque_data_digest = attach_opaque_data_digest;
+	error_context.conn_id_hex = conn_id_hex;
+	if (error_fields) {
+		error_context.fields = *error_fields;
+	}
+	error_context.fields[error_key::kUrl] = RedactCredentials(url);
+	error_context.fields.emplace(error_key::kRpcMethod, method_name);
 	auto response_body =
 	    HttpPostArrowIpc(context, url, body, auth, cookie_jar, cached_http_params, client_holder,
-	                     effective_caps, iroh_config);
+	                     effective_caps, iroh_config, /*cancellation=*/nullptr, &error_context);
 
 	// Parse the Arrow IPC response. Move the body in — the string becomes the
 	// owning Arrow buffer, avoiding an alloc+memcpy of the whole payload.

src/vgi_http_function_connection.cpp

@@ -56,6 +56,26 @@ static ErrorInfo HttpFunctionInfo(const char *subtype, const std::string &base_u
 	return info;
 }
 
+// Error context for a request whose worker error comes back in the HTTP body:
+// the worker (base URL), this execution, and the function. `url` is the
+// request route, reported as its own field.
+static RpcErrorContext HttpCallErrorContext(const std::string &base_url, const std::string &url,
+                                            const std::string &method, const std::string &function_name,
+                                            const std::string &function_type, const std::string &execution_id_hex,
+                                            const std::string &attach_opaque_data_digest,
+                                            const std::string &conn_id_hex) {
+	RpcErrorContext ctx;
+	ctx.worker_path = base_url;
+	ctx.execution_id_hex = execution_id_hex;
+	ctx.attach_opaque_data_digest = attach_opaque_data_digest;
+	ctx.conn_id_hex = conn_id_hex;
+	auto info = HttpFunctionInfo(nullptr, base_url, function_name, function_type);
+	info.erase(error_key::kErrorSubtype);
+	info.Url(url).Rpc(method);
+	ctx.fields.insert(info.begin(), info.end());
+	return ctx;
+}
+
 // Helper: parse ``vgi_batch_index`` off Arrow custom_metadata.
 // Returns INVALID_INDEX when the key is absent. Raises IOException
 // when present but un-parseable. The cap and monotonicity checks
@@ -423,10 +443,12 @@ BindResult HttpFunctionConnection::PerformBindRpc() {
 		// In/out capability snapshot: picks this request's Content-Encoding and
 		// is refreshed from the response (see CurrentCapabilities).
 		ServerCapabilities b_caps = CurrentCapabilities();
+		const auto call_error_fields = HttpFunctionInfo(nullptr, base_url_, function_name_, function_type_);
 		auto resp = HttpInvokeUnary(context_, base_url_, "bind", rpc_params, auth,
 		                             /*cookie_jar=*/nullptr, cached_params,
 		                             "", "", "", "", VGI_MAIN_PROTOCOL, &http_client_, &b_caps,
-		                             attach_params_ ? attach_params_->iroh() : nullptr);
+		                             attach_params_ ? attach_params_->iroh() : nullptr,
+		                             &call_error_fields);
 		PublishHarvestedCapabilities(b_caps);
 		if (!resp.batch || resp.batch->num_rows() == 0) {
 			throw IOException(HttpFunctionInfo(error_subtype::kProtocolViolation, base_url_, function_name_, function_type_).Rpc("bind"), "Empty bind response from HTTP server [url: %s]", base_url_);
@@ -578,10 +600,12 @@ InitResult HttpFunctionConnection::PerformInit(const BindResult &bind_result,
 	// codec is right on the first try (see CurrentCapabilities); refreshed from
 	// the response below.
 	ServerCapabilities harvested = CurrentCapabilities();
+	const auto call_error_context = HttpCallErrorContext(base_url_, init_url, "init", function_name_, function_type_,
+	                                                       GetExecutionIdHex(), GetAttachOpaqueDataDigest(), GetConnIdHex());
 	auto response_body = HttpPostArrowIpc(context_, init_url, body, auth,
 	                                        /*cookie_jar=*/nullptr, cached_params_init, &http_client_,
 	                                        &harvested, attach_params_ ? attach_params_->iroh() : nullptr,
-	                                        &context_.interrupted);
+	                                        &context_.interrupted, &call_error_context);
 	PublishHarvestedCapabilities(harvested);
 #ifdef __EMSCRIPTEN__
 #endif
@@ -806,11 +830,13 @@ HttpFunctionConnection::RpcTableBufferingProcess(const std::string &function_nam
 	auto cached_params = attach_params_ && IsHttpTransport(base_url_)
 	    ? attach_params_->GetOrInitHttpParams(context_, base_url_) : nullptr;
 	ServerCapabilities tb_caps = CurrentCapabilities();
+	const auto call_error_fields = HttpFunctionInfo(nullptr, base_url_, function_name_, function_type_);
 	auto resp = HttpInvokeUnary(context_, base_url_, "table_buffering_process", rpc_params, auth,
 	                             /*cookie_jar=*/nullptr, cached_params,
 	                             GetExecutionIdHex(), GetAttachOpaqueDataDigest(), "", GetConnIdHex(),
 	                             VGI_MAIN_PROTOCOL, &http_client_, &tb_caps,
-	                             attach_params_ ? attach_params_->iroh() : nullptr);
+	                             attach_params_ ? attach_params_->iroh() : nullptr,
+	                             &call_error_fields);
 	PublishHarvestedCapabilities(tb_caps);
 	auto inner = DecodeHttpOuterResponse(resp, "table_buffering_process", base_url_);
 	if (!inner || inner->num_rows() == 0) {
@@ -834,11 +860,13 @@ HttpFunctionConnection::RpcTableBufferingCombine(const std::string &function_nam
 	auto cached_params = attach_params_ && IsHttpTransport(base_url_)
 	    ? attach_params_->GetOrInitHttpParams(context_, base_url_) : nullptr;
 	ServerCapabilities tb_caps = CurrentCapabilities();
+	const auto call_error_fields = HttpFunctionInfo(nullptr, base_url_, function_name_, function_type_);
 	auto resp = HttpInvokeUnary(context_, base_url_, "table_buffering_combine", rpc_params, auth,
 	                             /*cookie_jar=*/nullptr, cached_params,
 	                             GetExecutionIdHex(), GetAttachOpaqueDataDigest(), "", GetConnIdHex(),
 	                             VGI_MAIN_PROTOCOL, &http_client_, &tb_caps,
-	                             attach_params_ ? attach_params_->iroh() : nullptr);
+	                             attach_params_ ? attach_params_->iroh() : nullptr,
+	                             &call_error_fields);
 	PublishHarvestedCapabilities(tb_caps);
 	auto inner = DecodeHttpOuterResponse(resp, "table_buffering_combine", base_url_);
 	if (!inner || inner->num_rows() == 0) {
@@ -868,11 +896,13 @@ void HttpFunctionConnection::RpcTableBufferingDestructor(const std::string &func
 	auto cached_params = attach_params_ && IsHttpTransport(base_url_)
 	    ? attach_params_->GetOrInitHttpParams(context_, base_url_) : nullptr;
 	ServerCapabilities tb_caps = CurrentCapabilities();
+	const auto call_error_fields = HttpFunctionInfo(nullptr, base_url_, function_name_, function_type_);
 	auto resp = HttpInvokeUnary(context_, base_url_, "table_buffering_destructor", rpc_params, auth,
 	                             /*cookie_jar=*/nullptr, cached_params,
 	                             GetExecutionIdHex(), GetAttachOpaqueDataDigest(), "", GetConnIdHex(),
 	                             VGI_MAIN_PROTOCOL, &http_client_, &tb_caps,
-	                             attach_params_ ? attach_params_->iroh() : nullptr);
+	                             attach_params_ ? attach_params_->iroh() : nullptr,
+	                             &call_error_fields);
 	PublishHarvestedCapabilities(tb_caps);
 	auto inner = DecodeHttpOuterResponse(resp, "table_buffering_destructor", base_url_);
 	(void)inner;
@@ -929,10 +959,12 @@ std::shared_ptr<arrow::RecordBatch> HttpFunctionConnection::ReadDataBatch() {
 		auto p_cached_params = attach_params_ && IsHttpTransport(exchange_url)
 		    ? attach_params_->GetOrInitHttpParams(context_, exchange_url) : nullptr;
 		ServerCapabilities p_harvested = CurrentCapabilities();
+		const auto call_error_context = HttpCallErrorContext(base_url_, exchange_url, "exchange", function_name_, function_type_,
+		                                                       GetExecutionIdHex(), GetAttachOpaqueDataDigest(), GetConnIdHex());
 		auto response_body = HttpPostArrowIpc(context_, exchange_url, body, p_auth,
 		                                        /*cookie_jar=*/nullptr, p_cached_params, &http_client_,
 		                                        &p_harvested, attach_params_ ? attach_params_->iroh() : nullptr,
-		                                        &context_.interrupted);
+		                                        &context_.interrupted, &call_error_context);
 		PublishHarvestedCapabilities(p_harvested);
 
 		// Parse response — buffer new data batches (adopt the body, no copy)
@@ -1007,10 +1039,12 @@ std::shared_ptr<arrow::RecordBatch> HttpFunctionConnection::ReadDataBatch() {
 	auto x_cached_params = attach_params_ && IsHttpTransport(exchange_url)
 	    ? attach_params_->GetOrInitHttpParams(context_, exchange_url) : nullptr;
 	ServerCapabilities x_harvested = CurrentCapabilities();
+	const auto call_error_context = HttpCallErrorContext(base_url_, exchange_url, "exchange", function_name_, function_type_,
+	                                                       GetExecutionIdHex(), GetAttachOpaqueDataDigest(), GetConnIdHex());
 	auto response_body = HttpPostArrowIpc(context_, exchange_url, body, x_auth,
 	                                        /*cookie_jar=*/nullptr, x_cached_params, &http_client_,
 	                                        &x_harvested, attach_params_ ? attach_params_->iroh() : nullptr,
-	                                        &context_.interrupted);
+	                                        &context_.interrupted, &call_error_context);
 	PublishHarvestedCapabilities(x_harvested);
 
 	// Parse response — copy into owning buffer since Arrow IPC reads zero-copy reference it