feat(cache): HTTP conditional revalidation via /init-request validators

Result-cache conditional revalidation (304 / not_modified) now works over
HTTP, not just subprocess. The validators (vgi.cache.if_none_match /
if_modified_since) rode the first producer tick, which reaches the worker
before it produces on subprocess. Over HTTP the first producer turn folds
into the /init request, so the tick arrived too late and the worker always
recomputed.

Fix: on the HTTP path, attach the validators to the /init request's
custom_metadata. SerializeRpcRequest gains an optional extra_metadata
param; HttpFunctionConnection::PerformInit passes the armed validators
(guarded by cond_sent_ for single delivery). Subprocess is unchanged
(keeps first-tick delivery). Detection/consumption is already
transport-agnostic — the worker's 0-row not_modified batch comes back in
the /init response buffer and flows through GetLastCacheControl →
MaybeSlideRevalidatedEntry → CachedReplayConnection swap, exactly as on
subprocess.

The worker side is handled in vgi-rpc (surfacing init-request metadata to
the producer's first process()); the vgi-python worker needs no change
(TableProducerState.process already reads the validators). revalidate.test
now `require httpfs` so it runs on HTTP; full cache suite green subprocess
(727) + HTTP (668, revalidate included).

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

Author: rustyconover

CLAUDE.md

@@ -581,13 +581,17 @@ log carries `tier=disk_streaming` vs `tier=memory` so the streaming path is test
 
 **Conditional revalidation (304).** A stale-but-`revalidatable` entry is probed by
 `LookupForRevalidation` *before* `Lookup` (which drops stale) — if the payload ≥
-`vgi_result_cache_revalidate_min_bytes`, the first producer tick carries `if_none_match` /
-`if_modified_since`. If the worker replies with a 0-row `not_modified` batch, `GetNextBatch` slides
-the entry's TTL and swaps to a `CachedReplayConnection` over the stored bytes (single-threaded, no
-re-stream); if it streams fresh data instead, the parallel capture commits a replacement.
-Revalidatable entries survive TTL reaping (refreshed on access, not dropped). **Subprocess only**
-(over HTTP the first producer exchange folds into `/init` before any tick, so the validator can't
-reach the worker's first `process()` — a follow-up).
+`vgi_result_cache_revalidate_min_bytes`, the client sends `if_none_match` / `if_modified_since`. If
+the worker replies with a 0-row `not_modified` batch, `GetNextBatch` slides the entry's TTL and swaps
+to a `CachedReplayConnection` over the stored bytes (single-threaded, no re-stream); if it streams
+fresh data instead, the parallel capture commits a replacement. Revalidatable entries survive TTL
+reaping (refreshed on access, not dropped). **Works on both transports.** Subprocess carries the
+validators on the first producer tick. Over HTTP the first producer turn folds into the `/init`
+request, so the C++ client attaches the validators to the `/init` request `custom_metadata`
+(`SerializeRpcRequest` `extra_metadata`) and the worker's framework (vgi-rpc `_run_http_producer_init`)
+surfaces that init-request metadata to the producer's first `process()` — so `not_modified` fires
+identically. Detection/consumption (`GetLastCacheControl` → `MaybeSlideRevalidatedEntry` → replay swap)
+is transport-agnostic (the 304 arrives in the `/init` response buffer over HTTP).
 
 **Files:** `src/vgi_result_cache.cpp` (singleton + disk tier + revalidation lookup),
 `src/vgi_cached_replay_connection.cpp` (serve), `src/vgi_result_cache_functions.cpp` (diagnostics),

src/include/vgi_rpc_client.hpp

@@ -7,6 +7,7 @@
 
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "duckdb/main/client_context.hpp"
@@ -158,9 +159,15 @@ StreamHeaderResult ReadStreamHeader(int fd, ClientContext *context,
 // `vgi_rpc.protocol_version` metadata instead of the global VGI_PROTOCOL_VERSION.
 // Used by the separately-versioned secret protocol (VGI_SECRET_PROTOCOL_VERSION);
 // all worker/catalog call sites leave it empty.
-std::vector<uint8_t> SerializeRpcRequest(const std::string &method_name,
-                                          const std::shared_ptr<arrow::RecordBatch> &params_batch,
-                                          const std::string &protocol_version_override = "");
+// extra_metadata: additional (key,value) pairs folded into the request's
+// custom_metadata alongside the method/version keys. Used by the HTTP transport to
+// carry the result-cache conditional-revalidation validators (vgi.cache.if_none_match
+// / if_modified_since) on the `/init` request, since over HTTP the worker's first
+// producer turn runs inside /init and must see the validators before it produces.
+std::vector<uint8_t> SerializeRpcRequest(
+    const std::string &method_name, const std::shared_ptr<arrow::RecordBatch> &params_batch,
+    const std::string &protocol_version_override = "",
+    const std::vector<std::pair<std::string, std::string>> &extra_metadata = {});
 
 // Serialize an RPC request with no parameters (zero-field schema, 1-row batch).
 std::vector<uint8_t> SerializeEmptyRpcRequest(const std::string &method_name);

src/vgi_http_function_connection.cpp

@@ -470,8 +470,24 @@ InitResult HttpFunctionConnection::PerformInit(const BindResult &bind_result,
 #ifdef __EMSCRIPTEN__
 #endif
 
-	// Serialize the init RPC request to Arrow IPC
-	auto body = SerializeRpcRequest("init", rpc_params);
+	// Serialize the init RPC request to Arrow IPC. Over HTTP the worker's first
+	// producer turn runs INSIDE this /init request (its output folds into the
+	// response), so a conditional-revalidation validator must ride the init request
+	// itself — the subprocess "first tick" arrives too late here. Attach
+	// vgi.cache.if_none_match / if_modified_since to the init request metadata when
+	// armed; the worker reads them off the init request and can answer 304
+	// not_modified in its first (init-time) batch. cond_sent_ guards single delivery.
+	std::vector<std::pair<std::string, std::string>> extra_metadata;
+	if (!cond_sent_ && (!cond_if_none_match_.empty() || !cond_if_modified_since_.empty())) {
+		if (!cond_if_none_match_.empty()) {
+			extra_metadata.emplace_back(VGI_CACHE_IF_NONE_MATCH_KEY, cond_if_none_match_);
+		}
+		if (!cond_if_modified_since_.empty()) {
+			extra_metadata.emplace_back(VGI_CACHE_IF_MODIFIED_SINCE_KEY, cond_if_modified_since_);
+		}
+		cond_sent_ = true;
+	}
+	auto body = SerializeRpcRequest("init", rpc_params, /*protocol_version_override=*/"", extra_metadata);
 #ifdef __EMSCRIPTEN__
 #endif
 

src/vgi_rpc_client.cpp

@@ -351,9 +351,10 @@ StreamHeaderResult ReadStreamHeader(int fd, ClientContext *context,
 // Buffer-based Serialization/Deserialization (for HTTP transport)
 // ============================================================================
 
-std::vector<uint8_t> SerializeRpcRequest(const std::string &method_name,
-                                          const std::shared_ptr<arrow::RecordBatch> &params_batch,
-                                          const std::string &protocol_version_override) {
+std::vector<uint8_t> SerializeRpcRequest(
+    const std::string &method_name, const std::shared_ptr<arrow::RecordBatch> &params_batch,
+    const std::string &protocol_version_override,
+    const std::vector<std::pair<std::string, std::string>> &extra_metadata) {
 	auto sink_result = arrow::io::BufferOutputStream::Create();
 	if (!sink_result.ok()) {
 		throw IOException("Failed to create buffer for RPC request: " + sink_result.status().ToString());
@@ -373,9 +374,14 @@ std::vector<uint8_t> SerializeRpcRequest(const std::string &method_name,
 	std::string protocol_version = protocol_version_override.empty()
 	                                   ? std::string(::duckdb::vgi::generated::VGI_PROTOCOL_VERSION)
 	                                   : protocol_version_override;
-	auto metadata = arrow::KeyValueMetadata::Make(
-	    {RPC_METHOD_KEY, RPC_REQUEST_VERSION_KEY, RPC_PROTOCOL_VERSION_KEY},
-	    {method_name, RPC_REQUEST_VERSION_VALUE, protocol_version});
+	std::vector<std::string> meta_keys = {RPC_METHOD_KEY, RPC_REQUEST_VERSION_KEY,
+	                                      RPC_PROTOCOL_VERSION_KEY};
+	std::vector<std::string> meta_values = {method_name, RPC_REQUEST_VERSION_VALUE, protocol_version};
+	for (const auto &kv : extra_metadata) {
+		meta_keys.push_back(kv.first);
+		meta_values.push_back(kv.second);
+	}
+	auto metadata = arrow::KeyValueMetadata::Make(std::move(meta_keys), std::move(meta_values));
 
 	auto status = writer->WriteRecordBatch(*params_batch, metadata);
 	if (!status.ok()) {

test/sql/integration/cache/revalidate.test

@@ -2,18 +2,19 @@
 # description: VGI result cache (M6) — conditional revalidation (304 / not_modified)
 # group: [vgi_integration_cache]
 #
-# Subprocess-only: the conditional-request validators ride the first producer
-# tick, which the worker's process() reads. Over HTTP the first producer exchange
-# is folded into the /init POST (its output is buffered before any tick is sent),
-# so the if_none_match validator never reaches the worker's first process() call
-# and the not_modified path can't fire. Omitting `require httpfs` makes this file
-# skip cleanly under the HTTP harness (the http:// ATTACH errors → auto-skip)
-# while still running on subprocess. See docs — HTTP revalidation is a follow-up.
+# Works on BOTH transports. The conditional-request validators
+# (vgi.cache.if_none_match / if_modified_since) ride the first producer tick on the
+# subprocess transport. Over HTTP the first producer turn is folded into the /init
+# request, so the C++ client attaches the validators to the /init request metadata
+# and the worker (via vgi_rpc) surfaces that metadata to its first process() call —
+# so the not_modified path fires identically over HTTP.
 
 require-env VGI_TEST_WORKER
 
 require vgi
 
+require httpfs
+
 statement ok
 CALL enable_logging(level := 'debug');