feat(cache/exchange): conditional revalidation (304) for streaming + LATERAL

Wire the producer cache's conditional-revalidation flow to the streaming table-in-out
(M1) and correlated-LATERAL (M2) exchange operators, both transports. A worker can now
advertise the always-revalidate contract (ttl=0 + etag + revalidatable) on exchange
output: the entry is stored immediately-stale (memory-only), and a repeat scan confirms
freshness with a cheap 0-row not_modified reply instead of recomputing.

- Operators: probe LookupForRevalidation FIRST (plain Lookup evicts a stale entry); if
  the entry is >= vgi_result_cache_revalidate_min_bytes, arm the exchange via
  SetConditionalRequest. After ReadDataBatch, a 0-row not_modified reply slides the
  entry's TTL (SlideRevalidatedExchangeEntry) and serves the stored bytes
  (result_cache.revalidate outcome=not_modified); a fresh reply recaptures.
- Connection: FunctionConnection::WriteInputBatch now attaches the
  vgi.cache.if_none_match/if_modified_since validators to the exchange input batch's
  metadata (the HTTP path already did via SerializeBatchWithState).
- StoreExchangeMemoEntry accepts the always-revalidate (ttl=0 + etag + revalidatable)
  entry, storing it immediately-stale + memory-only (LookupForRevalidation probes memory;
  a disk immediately-stale blob is un-loadable). SlideRevalidatedExchangeEntry keeps an
  always-revalidate entry immediately-stale on slide (doesn't fall to a positive default).

NOT wired for buffered (M3): its combine/finalize request model (key known only at
Finalize) doesn't fit the per-unit validator flow — documented follow-up.

Test: cache/exchange_revalidate.test (both transports, 45 assertions — M1 + M2 304). Full
cache suite green (916 assertions); function_registration 132 -> 134.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XZMDnzigtwqubFrqbP8JEE

Author: rustyconover

CLAUDE.md

@@ -674,10 +674,25 @@ by default (needs `vgi_result_cache_dir`), so this only persists when configured
 `BuildExchangeCacheKeyStatic` calls `SyncResultCacheSettings` (mirroring the producer's
 `ConfigureIfChanged`) so `SET vgi_result_cache_dir/…` reaches the singleton on the exchange
 path; `DeserializeCachedRecordBatch` is disk-aware (positioned-reads a streaming entry's
-batch from the blob). Transaction scope is refused for exchange entries in v1. Conditional
-revalidation (etag/last-modified/304/stale-while-revalidate) is **producer-only** — exchange
-entries store the validator fields but a stale entry is a plain miss, not a 304 refresh.
-Fixtures (vgi-python `_test_fixtures/table_in_out.py`):
+batch from the blob). Transaction scope is refused for exchange entries in v1.
+
+**Conditional revalidation (304).** Wired for the **streaming table-in-out (M1)** and
+**correlated LATERAL (M2)** operators (both transports). A worker advertises the
+always-revalidate contract (`ttl=0 + etag + revalidatable` → stored but immediately
+stale, memory-only). On a repeat, the operator probes `LookupForRevalidation` FIRST
+(plain `Lookup` evicts a stale entry), and if the entry is ≥
+`vgi_result_cache_revalidate_min_bytes` arms the exchange: the stored validators ride
+the `WriteInputBatch` input-batch metadata (`SetConditionalRequest` →
+`vgi.cache.if_none_match`/`if_modified_since`), the worker's exchange `process()` reads
+them off `input.custom_metadata` (surfaced on `ProcessParams`) and answers a 0-row
+`CacheControl(not_modified=True)`, and the operator slides the entry's TTL
+(`SlideRevalidatedExchangeEntry`) + serves the stored bytes (`result_cache.revalidate
+outcome=not_modified`) instead of recomputing. The vgi-python buffered/exchange
+`finalize()`/`exchange()` now surface the validators. **NOT wired for buffered (M3)** —
+its request model (combine/finalize, key known only at Finalize) doesn't fit the
+per-unit validator flow; a buffered stale entry is a plain miss (documented follow-up).
+Fixtures: `cached_reval_echo` (M1 classic), `cached_reval_double` (M2 blended).
+Other fixtures (vgi-python `_test_fixtures/table_in_out.py`):
 `cached_echo` (streaming), `cached_double` (LATERAL), `cached_sum_all` (buffered). Tests:
 `test/sql/integration/cache/exchange_{streaming,lateral,buffered}.test` (both transports;
 hit-skips-worker, LATERAL order-independence + correlated correctness, shared surface + flush).

src/include/vgi_exchange_cache_key.hpp

@@ -104,6 +104,12 @@ struct ExchangeStoreResult {
 	int64_t bytes = 0;
 };
 
+//! On a 304 not_modified reply, re-insert the stored entry with a slid TTL (and any
+//! refreshed validators from `cc`) so future lookups hit fresh without re-fetching.
+//! Mirrors the producer's MaybeSlideRevalidatedEntry. Best-effort.
+void SlideRevalidatedExchangeEntry(const VgiResultCacheEntry &entry, const VgiCacheControl &cc,
+                                   int64_t default_ttl_seconds, bool allow_disk);
+
 //! Build a cache entry from one input unit's output batches and Insert it. Freshness
 //! from `cc`: a positive ttl (from cc.ttl_seconds or default_ttl_seconds) is required;
 //! no_store and transaction-scoped results are refused. `allow_disk` opts the entry

src/include/vgi_table_in_out_impl.hpp

@@ -218,6 +218,9 @@ struct VgiTableInOutGlobalState : public GlobalTableFunctionState {
 	VgiResultCacheKey cache_key;             // static portion; input_hash set per exchange
 	std::string cache_catalog_name;
 	int64_t cache_default_ttl_seconds = 0;
+	// Min stored-payload size before a stale revalidatable entry is conditionally
+	// revalidated (below it, refetch instead of a conditional request).
+	int64_t cache_revalidate_min_bytes = 262144;
 };
 
 // ============================================================================

src/vgi_exchange_cache_key.cpp

@@ -313,6 +313,38 @@ std::shared_ptr<arrow::RecordBatch> DeserializeCachedRecordBatch(const VgiResult
 	return next_result.ValueUnsafe().batch;
 }
 
+void SlideRevalidatedExchangeEntry(const VgiResultCacheEntry &entry, const VgiCacheControl &cc,
+                                   int64_t /*default_ttl_seconds*/, bool allow_disk) {
+	auto fresh = std::make_shared<VgiResultCacheEntry>(entry); // shallow copy (shared Buffers)
+	fresh->stored_at = std::chrono::steady_clock::now();
+	// A fresh ttl on the not_modified batch wins; else reuse the PRIOR lifetime so a
+	// validator-only 304 slides forward — and, crucially, an always-revalidate entry
+	// (prior lifetime 0 / immediately stale) stays immediately stale so it keeps
+	// revalidating (do NOT fall to a positive default). Mirrors the producer.
+	int64_t ttl = cc.ttl_seconds.value_or(0);
+	if (ttl <= 0 && !entry.never_expires) {
+		auto prior =
+		    std::chrono::duration_cast<std::chrono::seconds>(entry.expires_at - entry.stored_at).count();
+		ttl = prior > 0 ? prior : 0;
+	}
+	if (ttl > VGI_CACHE_MAX_TTL_SECONDS) {
+		ttl = VGI_CACHE_MAX_TTL_SECONDS;
+	}
+	if (!entry.never_expires) {
+		fresh->expires_at = fresh->stored_at + std::chrono::seconds(ttl > 0 ? ttl : 0);
+	}
+	if (!cc.etag.empty()) {
+		fresh->etag = cc.etag;
+	}
+	if (!cc.last_modified.empty()) {
+		fresh->last_modified = cc.last_modified;
+	}
+	// An immediately-stale (always-revalidate) entry is memory-only: LookupForRevalidation
+	// probes the in-memory index, and a disk immediately-stale blob is un-loadable.
+	bool eff_allow_disk = allow_disk && (entry.never_expires || fresh->expires_at > fresh->stored_at);
+	VgiResultCache::Instance().Insert(fresh, eff_allow_disk);
+}
+
 ExchangeStoreResult StoreExchangeMemoEntry(const VgiResultCacheKey &key, const VgiCacheControl &cc,
                                            const std::string &catalog_name, int64_t default_ttl_seconds,
                                            const std::vector<std::shared_ptr<arrow::RecordBatch>> &out_batches,
@@ -335,11 +367,17 @@ ExchangeStoreResult StoreExchangeMemoEntry(const VgiResultCacheKey &key, const V
 	if (ttl > VGI_CACHE_MAX_TTL_SECONDS) {
 		ttl = VGI_CACHE_MAX_TTL_SECONDS;
 	}
-	// v1: a positive ttl is required (expires-only advertisements fall back to the
-	// default; if that is also 0 there is no freshness basis → refuse).
-	if (ttl <= 0) {
+	// A positive ttl is required UNLESS this is the "always-revalidate" (HTTP no-cache)
+	// contract: ttl=0 + a validator (etag) + revalidatable — stored but immediately
+	// stale, so every read revalidates via a conditional request and a 304 reuses the
+	// stored bytes. Such an entry is memory-only (LookupForRevalidation probes memory;
+	// a disk immediately-stale blob is un-loadable).
+	const bool immediately_stale = ttl <= 0;
+	const bool revalidatable_no_cache = immediately_stale && cc.revalidatable && !cc.etag.empty();
+	if (immediately_stale && !revalidatable_no_cache) {
 		return skip("no_freshness");
 	}
+	const bool eff_allow_disk = allow_disk && !immediately_stale;
 
 	auto entry = std::make_shared<VgiResultCacheEntry>();
 	entry->key = key;
@@ -349,7 +387,7 @@ ExchangeStoreResult StoreExchangeMemoEntry(const VgiResultCacheKey &key, const V
 	entry->last_modified = cc.last_modified;
 	entry->revalidatable = cc.revalidatable;
 	entry->stored_at = std::chrono::steady_clock::now();
-	entry->expires_at = entry->stored_at + std::chrono::seconds(ttl);
+	entry->expires_at = entry->stored_at + std::chrono::seconds(immediately_stale ? 0 : ttl);
 
 	CachedStream stream;
 	int64_t rows = 0;
@@ -372,7 +410,7 @@ ExchangeStoreResult StoreExchangeMemoEntry(const VgiResultCacheKey &key, const V
 	entry->rows = rows;
 	entry->total_bytes = bytes;
 
-	if (!VgiResultCache::Instance().Insert(std::move(entry), allow_disk)) {
+	if (!VgiResultCache::Instance().Insert(std::move(entry), eff_allow_disk)) {
 		return skip("too_large_for_memory");
 	}
 	res.stored = true;

src/vgi_function_connection.cpp

@@ -1342,6 +1342,31 @@ void FunctionConnection::WriteInputBatch(const std::shared_ptr<arrow::RecordBatc
 	}
 #endif
 
+	// Conditional-revalidation validators (exchange-mode result cache): ride the
+	// exchange input batch's custom_metadata (the worker's exchange process() reads
+	// them, mirroring the producer's first tick) so the worker can answer 304-style
+	// with a 0-row vgi.cache.not_modified batch. The caller (an exchange operator)
+	// sets them per input unit via SetConditionalRequest and clears them after, so
+	// they never leak to the next unit. Merged with any shm pointer metadata.
+	if (!cond_if_none_match_.empty() || !cond_if_modified_since_.empty()) {
+		std::vector<std::string> keys, vals;
+		if (write_meta) {
+			for (int64_t i = 0; i < write_meta->size(); i++) {
+				keys.push_back(write_meta->key(i));
+				vals.push_back(write_meta->value(i));
+			}
+		}
+		if (!cond_if_none_match_.empty()) {
+			keys.emplace_back(VGI_CACHE_IF_NONE_MATCH_KEY);
+			vals.push_back(cond_if_none_match_);
+		}
+		if (!cond_if_modified_since_.empty()) {
+			keys.emplace_back(VGI_CACHE_IF_MODIFIED_SINCE_KEY);
+			vals.push_back(cond_if_modified_since_);
+		}
+		write_meta = arrow::KeyValueMetadata::Make(std::move(keys), std::move(vals));
+	}
+
 	auto write_status = write_meta ? input_writer_->WriteRecordBatch(*to_write, write_meta)
 	                               : input_writer_->WriteRecordBatch(*to_write);
 	if (!write_status.ok()) {

src/vgi_lateral_batch_operator.cpp

@@ -127,6 +127,7 @@ struct VgiLateralBatchOperatorState : public OperatorState {
 	VgiResultCacheKey cache_static_key;
 	std::string cache_catalog_name;
 	int64_t cache_default_ttl_seconds = 0;
+	int64_t cache_revalidate_min_bytes = 262144;
 	VgiCacheControl cache_cc;   // latched from the first exchange output
 	bool cache_cc_latched = false;
 	// MISS capture: the current input chunk's full key + its accumulated POST-STAMP
@@ -328,6 +329,10 @@ OperatorResultType PhysicalVgiLateralBatch::Execute(ExecutionContext &context, D
 			if (client_context.TryGetCurrentSetting("vgi_result_cache_default_ttl_seconds", ttl_v)) {
 				state.cache_default_ttl_seconds = static_cast<int64_t>(ttl_v.GetValue<uint64_t>());
 			}
+			Value rmv;
+			if (client_context.TryGetCurrentSetting("vgi_result_cache_revalidate_min_bytes", rmv) && !rmv.IsNull()) {
+				state.cache_revalidate_min_bytes = static_cast<int64_t>(rmv.GetValue<uint64_t>());
+			}
 		} else if (reason) {
 			VGI_LOG(client_context, "result_cache.ineligible",
 			        {{"function", bd.function_name}, {"reason", reason}});
@@ -377,20 +382,31 @@ OperatorResultType PhysicalVgiLateralBatch::Execute(ExecutionContext &context, D
 	// (B) Fresh input chunk. Cache lookup on the static key + order-independent
 	// FULL-chunk input hash. A HIT replays the cached POST-STAMP output (correlated
 	// columns baked in — no re-stamp; sound because the operator is NO_ORDER).
+	std::shared_ptr<const VgiResultCacheEntry> reval_entry; // armed conditional revalidation
 	if (state.cache_eligible) {
 		state.capture_key = state.cache_static_key;
 		state.capture_key.input_hash = HashInputChunkUnordered(client_context, input);
-		auto entry = VgiResultCache::Instance().Lookup(state.capture_key, std::chrono::steady_clock::now());
-		if (entry) {
-			VGI_LOG(client_context, "result_cache.hit",
-			        {{"function", bd.function_name}, {"key_hash", state.capture_key.HexDigest()}, {"tier", "memory"}});
-			if (entry->streams.empty() || entry->streams[0].batches.empty()) {
-				chunk.SetCardinality(0); // cached empty (all-filtered) result — nothing to emit
-				return OperatorResultType::NEED_MORE_INPUT;
+		auto now = std::chrono::steady_clock::now();
+		// Probe conditional revalidation FIRST (Lookup evicts a stale entry). If a
+		// large-enough stale revalidatable entry exists, arm the exchange with its
+		// validators; the worker confirms (304 → reuse stored) or returns fresh data.
+		auto reval = VgiResultCache::Instance().LookupForRevalidation(state.capture_key, now);
+		if (reval && reval->revalidatable && !reval->streams.empty() &&
+		    reval->total_bytes >= state.cache_revalidate_min_bytes) {
+			reval_entry = reval;
+		} else {
+			auto entry = VgiResultCache::Instance().Lookup(state.capture_key, now);
+			if (entry) {
+				VGI_LOG(client_context, "result_cache.hit",
+				        {{"function", bd.function_name}, {"key_hash", state.capture_key.HexDigest()}, {"tier", "memory"}});
+				if (entry->streams.empty() || entry->streams[0].batches.empty()) {
+					chunk.SetCardinality(0); // cached empty (all-filtered) result — nothing to emit
+					return OperatorResultType::NEED_MORE_INPUT;
+				}
+				state.serving = entry;
+				state.serve_cursor = 0;
+				return EmitServedSlice(state, state.serve_arrow_table, chunk);
 			}
-			state.serving = entry;
-			state.serve_cursor = 0;
-			return EmitServedSlice(state, state.serve_arrow_table, chunk);
 		}
 	}
 
@@ -408,6 +424,12 @@ OperatorResultType PhysicalVgiLateralBatch::Execute(ExecutionContext &context, D
 		state.connection = AcquireBlendedInputConnection(client_context, bd, state.substream_id, projection_ids);
 	}
 
+	// Arm conditional-revalidation validators for this chunk's exchange (they ride the
+	// input batch's metadata). Cleared right after the exchange so they don't leak.
+	if (reval_entry) {
+		state.connection->SetConditionalRequest(reval_entry->etag, reval_entry->last_modified);
+	}
+
 	// Split off the worker-input columns [0, input_length) as a zero-copy view.
 	DataChunk worker_input;
 	vector<LogicalType> in_types;
@@ -442,6 +464,28 @@ OperatorResultType PhysicalVgiLateralBatch::Execute(ExecutionContext &context, D
 		                  bd.worker_path(), bd.function_name);
 	}
 
+	// Conditional revalidation: clear the armed validators (so the next chunk's exchange
+	// isn't a stray conditional request) and, on a 304 (0-row not_modified reply), slide
+	// the stored entry's TTL and serve its cached POST-STAMP output instead of the worker
+	// response. Otherwise the fresh response falls through to the normal capture path.
+	if (reval_entry) {
+		state.connection->SetConditionalRequest("", "");
+		if (output_batch->num_rows() == 0) {
+			auto cc = state.connection->GetLastCacheControl();
+			if (cc.not_modified) {
+				SlideRevalidatedExchangeEntry(*reval_entry, cc, state.cache_default_ttl_seconds,
+				                              /*allow_disk=*/true);
+				VGI_LOG(client_context, "result_cache.revalidate",
+				        {{"function", bd.function_name},
+				         {"key_hash", state.capture_key.HexDigest()},
+				         {"outcome", "not_modified"}});
+				state.serving = reval_entry;
+				state.serve_cursor = 0;
+				return EmitServedSlice(state, state.serve_arrow_table, chunk);
+			}
+		}
+	}
+
 	// Latch the worker's cache-control advertisement off the first exchange output.
 	if (state.cache_eligible && !state.cache_cc_latched) {
 		state.cache_cc = state.connection->GetLastCacheControl();

src/vgi_table_in_out_impl.cpp

@@ -484,6 +484,10 @@ unique_ptr<GlobalTableFunctionState> VgiTableInOutInitGlobal(ClientContext &cont
 			if (context.TryGetCurrentSetting("vgi_result_cache_default_ttl_seconds", ttl_v)) {
 				global_state->cache_default_ttl_seconds = static_cast<int64_t>(ttl_v.GetValue<uint64_t>());
 			}
+			Value rmv;
+			if (context.TryGetCurrentSetting("vgi_result_cache_revalidate_min_bytes", rmv) && !rmv.IsNull()) {
+				global_state->cache_revalidate_min_bytes = static_cast<int64_t>(rmv.GetValue<uint64_t>());
+			}
 		} else if (reason) {
 			VGI_LOG(context, "result_cache.ineligible",
 			        {{"function", bind_data.function_name}, {"reason", reason}});
@@ -885,17 +889,32 @@ OperatorResultType VgiTableInOutFunction(ExecutionContext &context, TableFunctio
 	std::shared_ptr<arrow::RecordBatch> output_batch;
 	bool cache_hit = false;
 	VgiResultCacheKey batch_key;
+	// Conditional revalidation: a stale-but-revalidatable entry the worker can confirm
+	// via a cheap 0-row not_modified reply (validators ride the exchange input batch).
+	std::shared_ptr<const VgiResultCacheEntry> reval_entry;
 	if (global_state.cache_eligible) {
 		batch_key = global_state.cache_key;
 		batch_key.input_hash = HashInputBatchOrdered(input_batch);
-		auto entry = VgiResultCache::Instance().Lookup(batch_key, std::chrono::steady_clock::now());
-		if (entry && !entry->streams.empty() && !entry->streams[0].batches.empty()) {
-			output_batch = DeserializeCachedRecordBatch(*entry, entry->streams[0].batches[0]);
-			cache_hit = true;
-			VGI_LOG(client_context, "result_cache.hit",
-			        {{"function", bind_data.function_name},
-			         {"key_hash", batch_key.HexDigest()},
-			         {"tier", "memory"}});
+		auto now = std::chrono::steady_clock::now();
+		// Probe conditional revalidation FIRST — Lookup() drops (evicts) a stale entry,
+		// so calling it first would destroy the very entry we want to revalidate.
+		auto reval = VgiResultCache::Instance().LookupForRevalidation(batch_key, now);
+		if (reval && reval->revalidatable && !reval->streams.empty() &&
+		    reval->total_bytes >= global_state.cache_revalidate_min_bytes) {
+			// Arm the exchange with its validators; do NOT serve yet — the worker
+			// confirms (304 → reuse stored) or returns fresh data (recapture).
+			reval_entry = reval;
+			conn.SetConditionalRequest(reval->etag, reval->last_modified);
+		} else {
+			auto entry = VgiResultCache::Instance().Lookup(batch_key, now);
+			if (entry && !entry->streams.empty() && !entry->streams[0].batches.empty()) {
+				output_batch = DeserializeCachedRecordBatch(*entry, entry->streams[0].batches[0]);
+				cache_hit = true;
+				VGI_LOG(client_context, "result_cache.hit",
+				        {{"function", bind_data.function_name},
+				         {"key_hash", batch_key.HexDigest()},
+				         {"tier", "memory"}});
+			}
 		}
 	}
 
@@ -933,10 +952,33 @@ OperatorResultType VgiTableInOutFunction(ExecutionContext &context, TableFunctio
 			output_batch = conn.ReadDataBatch();
 		}
 
+		// Conditional revalidation: if we armed validators for this unit, clear them so
+		// the next input batch's exchange isn't a stray conditional request. If the
+		// worker answered 304 (a 0-row vgi.cache.not_modified batch), slide the stored
+		// entry's TTL and serve its bytes instead of re-storing — the payload is
+		// unchanged. Otherwise the fresh response falls through to normal capture below
+		// (replacing the stale entry).
+		if (reval_entry) {
+			conn.SetConditionalRequest("", "");
+			if (output_batch && output_batch->num_rows() == 0) {
+				auto cc = conn.GetLastCacheControl();
+				if (cc.not_modified) {
+					SlideRevalidatedExchangeEntry(*reval_entry, cc, global_state.cache_default_ttl_seconds,
+					                              /*allow_disk=*/true);
+					output_batch = DeserializeCachedRecordBatch(*reval_entry, reval_entry->streams[0].batches[0]);
+					cache_hit = true; // serve the stored bytes; skip the store below
+					VGI_LOG(client_context, "result_cache.revalidate",
+					        {{"function", bind_data.function_name},
+					         {"key_hash", batch_key.HexDigest()},
+					         {"outcome", "not_modified"}});
+				}
+			}
+		}
+
 		// Latch the worker's cache-control advertisement off the first exchange
 		// output, then (if cacheable) memoize this input batch's output. Skip on the
 		// terminal EOS batch (nullptr) — there is nothing to cache and cc rides data.
-		if (global_state.cache_eligible && output_batch) {
+		if (!cache_hit && global_state.cache_eligible && output_batch) {
 			if (!local_state.cache_cc_latched) {
 				local_state.cache_cc = conn.GetLastCacheControl();
 				local_state.cache_cc_latched = true;