buffering: make source producers HTTP-state-serializable

Buffering source/finalize producers held a live BoundStorage (SQLite-backed)
view in their StreamState. Over HTTP that state is serialized into a
continuation token between /init and /exchange, and a DB connection can't be
serialized (infinite recursion). Keep the storage view transient and re-acquire
it from (executionId, attachId) on resume:

- new BufferingStorageHolder: process-wide handle to the worker's FunctionStorage
  backend, registered by VgiServiceImpl; bind(executionId, attachId) re-creates a
  BoundStorage.
- BufferingFinalizeProducer: storage view is transient; executionId/attachId/
  finalizeStateId are serialized; storage() lazily re-binds after a round-trip;
  adds a no-arg ctor.
- thread attachId through TableBufferingFinalizeParams.
- the 7 finalize-producer subclasses: no-arg ctor, storage() accessor, non-final
  cursor fields so the reflection-based serializer can repopulate them.

No behaviour change on the in-memory transports (storage view is set at init and
storage() returns it). Brings the over-http suite from 30 -> 7 table-function
failures (with the framework state-serializer fixes).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: rustyconover

vgi-example-worker/src/main/java/farm/query/vgi/example/accumulate/AccumulateFunction.java

@@ -166,12 +166,14 @@ public TableProducerState createFinalizeProducer(TableBufferingFinalizeParams pa
     private static final class OutDrainProducer extends BufferingFinalizeProducer {
         private long afterId = 0;
 
+        private OutDrainProducer() {}
+
         OutDrainProducer(TableBufferingFinalizeParams params) {
             super(params);
         }
 
         @Override public void produceTick(OutputCollector out, CallContext ctx) {
-            List<FunctionStorage.LogEntry> rows = storage.stateLogScan(NS_OUT, KEY, afterId, 1);
+            List<FunctionStorage.LogEntry> rows = storage().stateLogScan(NS_OUT, KEY, afterId, 1);
             if (rows.isEmpty()) {
                 out.finish();
                 return;

vgi-example-worker/src/main/java/farm/query/vgi/example/buffering/AbstractBufferAndDrain.java

@@ -58,16 +58,18 @@ public abstract class AbstractBufferAndDrain implements TableBufferingFunction {
 
     /** Drains a {@code (ns, "")} state-log, one buffered batch per tick. */
     static final class LogDrainProducer extends BufferingFinalizeProducer {
-        private final byte[] ns;
+        private byte[] ns;
         private long afterId = -1;
 
+        LogDrainProducer() {}
+
         LogDrainProducer(TableBufferingFinalizeParams params, byte[] ns) {
             super(params);
             this.ns = ns;
         }
 
         @Override public void produceTick(OutputCollector out, CallContext ctx) {
-            List<FunctionStorage.LogEntry> rows = storage.stateLogScan(ns, KEY, afterId, 1);
+            List<FunctionStorage.LogEntry> rows = storage().stateLogScan(ns, KEY, afterId, 1);
             if (rows.isEmpty()) {
                 out.finish();
                 return;

vgi-example-worker/src/main/java/farm/query/vgi/example/buffering/BufferEmitWideFunction.java

@@ -67,9 +67,11 @@ public final class BufferEmitWideFunction implements TableBufferingFunction {
     }
 
     private static final class WideProducer extends BufferingFinalizeProducer {
-        private final long rows;
+        private long rows;
         private boolean emitted = false;
 
+        private WideProducer() {}
+
         WideProducer(TableBufferingFinalizeParams params, long rows) {
             super(params);
             this.rows = rows;

vgi-example-worker/src/main/java/farm/query/vgi/example/buffering/LargeStateFunction.java

@@ -55,12 +55,14 @@ public final class LargeStateFunction extends AbstractBufferAndDrain {
     private static final class TotalSizeProducer extends BufferingFinalizeProducer {
         private boolean emitted = false;
 
+        private TotalSizeProducer() {}
+
         TotalSizeProducer(TableBufferingFinalizeParams params) { super(params); }
 
         @Override public void produceTick(OutputCollector out, CallContext ctx) {
             if (emitted) { out.finish(); return; }
             long total = 0;
-            for (FunctionStorage.LogEntry e : storage.stateLogScan(NS_LARGE, KEY, -1, Integer.MAX_VALUE)) {
+            for (FunctionStorage.LogEntry e : storage().stateLogScan(NS_LARGE, KEY, -1, Integer.MAX_VALUE)) {
                 total += e.value().length;
             }
             List<FieldVector> vectors = new ArrayList<>();

vgi-example-worker/src/main/java/farm/query/vgi/example/buffering/OrderedSourceFunction.java

@@ -70,9 +70,11 @@ public final class OrderedSourceFunction implements TableBufferingFunction {
 
     /** Emits exactly one row carrying the integer decoded from finalize_state_id. */
     private static final class OneShotProducer extends BufferingFinalizeProducer {
-        private final long value;
+        private long value;
         private boolean emitted = false;
 
+        private OneShotProducer() {}
+
         OneShotProducer(TableBufferingFinalizeParams params) {
             super(params);
             this.value = new BigInteger(1, params.finalizeStateId()).longValue();

vgi-example-worker/src/main/java/farm/query/vgi/example/buffering/SlowCancellableBufferingFunction.java

@@ -65,10 +65,12 @@ public final class SlowCancellableBufferingFunction implements TableBufferingFun
     }
 
     private static final class SlowProducer extends BufferingFinalizeProducer {
-        private final long sleepMs;
-        private final long total;
+        private long sleepMs;
+        private long total;
         private long emitted = 0;
 
+        private SlowProducer() {}
+
         SlowProducer(TableBufferingFinalizeParams params, long sleepMs, long total) {
             super(params);
             this.sleepMs = sleepMs;

vgi-example-worker/src/main/java/farm/query/vgi/example/buffering/SumAllColumnsBufferingFunction.java

@@ -103,6 +103,8 @@ public class SumAllColumnsBufferingFunction extends AbstractBufferAndDrain {
     private static final class SumProducer extends BufferingFinalizeProducer {
         private boolean emitted = false;
 
+        private SumProducer() {}
+
         SumProducer(TableBufferingFinalizeParams params) { super(params); }
 
         @Override public void produceTick(OutputCollector out, CallContext ctx) {
@@ -117,7 +119,7 @@ private static final class SumProducer extends BufferingFinalizeProducer {
                 if (dst instanceof BigIntVector bi) bi.setSafe(0, 0L);
                 else if (dst instanceof Float8Vector fl) fl.setSafe(0, 0.0);
             }
-            for (FunctionStorage.LogEntry e : storage.stateLogScan(NS_RAW, KEY, -1, Integer.MAX_VALUE)) {
+            for (FunctionStorage.LogEntry e : storage().stateLogScan(NS_RAW, KEY, -1, Integer.MAX_VALUE)) {
                 try (VectorSchemaRoot src = BatchUtil.readSingleBatch(e.value(), Allocators.root())) {
                     int rows = src.getRowCount();
                     for (Field f : outputSchema.getFields()) {

vgi/src/main/java/farm/query/vgi/buffering/BufferingFinalizeProducer.java

@@ -26,10 +26,20 @@
  */
 public abstract class BufferingFinalizeProducer extends TableProducerState {
 
-    /** Read view over the batches buffered during the Sink phase. */
-    protected final BoundStorage storage;
+    /** Read view over the batches buffered during the Sink phase. Transient: a
+     *  live storage view (SQLite connection) can't be serialized into an HTTP
+     *  state token, so it's re-acquired from {@link #executionId}/{@link #attachId}
+     *  via {@link #storage()} when this producer is resumed on /exchange. */
+    protected transient BoundStorage storageView;
+    /** Execution id the storage is scoped to (survives the state token). */
+    protected byte[] executionId;
+    /** Attach plaintext for per-attach shard routing (survives the state token). */
+    protected byte[] attachId;
     /** State id selecting which combined partition this producer drains. */
-    protected final byte[] finalizeStateId;
+    protected byte[] finalizeStateId;
+
+    /** No-arg constructor for HTTP state-token deserialization. */
+    protected BufferingFinalizeProducer() {}
 
     /**
      * Captures the storage view + {@code finalize_state_id} and forwards the
@@ -39,10 +49,24 @@ public abstract class BufferingFinalizeProducer extends TableProducerState {
      */
     protected BufferingFinalizeProducer(TableBufferingFinalizeParams params) {
         super(params.initParams());
-        this.storage = params.storage();
+        this.storageView = params.storage();
+        this.executionId = params.executionId();
+        this.attachId = params.attachId();
         this.finalizeStateId = params.finalizeStateId();
     }
 
+    /**
+     * The storage view, re-acquired from {@code (executionId, attachId)} if this
+     * producer was resumed from an HTTP state token (the transient view is null
+     * after deserialization).
+     *
+     * @return the per-execution storage view
+     */
+    protected BoundStorage storage() {
+        if (storageView == null) storageView = BufferingStorageHolder.bind(executionId, attachId);
+        return storageView;
+    }
+
     /**
      * Narrows {@code full} to {@link #outputSchema} (by field name), applies
      * pushdown filters, and emits the result. The emitted root's vectors are

vgi/src/main/java/farm/query/vgi/buffering/BufferingStorageHolder.java

@@ -0,0 +1,49 @@
+// Copyright 2026 Query Farm LLC - https://query.farm
+
+package farm.query.vgi.buffering;
+
+import farm.query.vgi.storage.BoundStorage;
+import farm.query.vgi.storage.FunctionStorage;
+
+/**
+ * Process-wide handle to the worker's {@link FunctionStorage} backend, so a
+ * buffering source producer can re-bind its {@link BoundStorage} after an HTTP
+ * state-token round-trip. The HTTP transport is stateless: it serializes a
+ * producer's state between {@code /init} and {@code /exchange}, and a live
+ * storage view (a SQLite connection) can't be serialized. So buffering source
+ * producers keep their {@code storage} view {@code transient} and re-acquire it
+ * here from the {@code (executionId, attachId)} they did serialize.
+ *
+ * <p>A single worker process serves one storage backend, so a static handle is
+ * sufficient; {@code VgiServiceImpl} registers it at construction.</p>
+ */
+public final class BufferingStorageHolder {
+
+    private static volatile FunctionStorage backend;
+
+    private BufferingStorageHolder() {}
+
+    /**
+     * Register the worker's storage backend. Called once when the service is built.
+     *
+     * @param b the worker's {@link FunctionStorage} backend
+     */
+    public static void register(FunctionStorage b) { backend = b; }
+
+    /**
+     * Re-bind a {@link BoundStorage} for {@code executionId} on the registered
+     * backend (used when resuming a buffering source producer from a state token).
+     *
+     * @param executionId   the buffering execution id the storage is scoped to
+     * @param attachId      the attach plaintext (for per-attach shard routing), or {@code null}
+     * @return a storage view bound to {@code executionId}
+     * @throws IllegalStateException if no backend has been registered
+     */
+    public static BoundStorage bind(byte[] executionId, byte[] attachId) {
+        FunctionStorage b = backend;
+        if (b == null) {
+            throw new IllegalStateException("no buffering storage backend registered");
+        }
+        return new BoundStorage(b, executionId, attachId);
+    }
+}

vgi/src/main/java/farm/query/vgi/buffering/TableBufferingFinalizeParams.java

@@ -14,11 +14,13 @@
  *
  * @param executionId the opaque {@code execution_id} identifying this buffering execution.
  * @param finalizeStateId the opaque {@code finalize_state_id} naming the output stream this producer drains.
+ * @param attachId the attach plaintext for per-attach shard routing (lets a source producer re-bind storage after an HTTP state-token round-trip).
  * @param storage the storage view bound to {@code executionId}, for reading back stashed state.
  * @param initParams the table-init parameters carrying the projection-narrowed output schema and pushdown filter.
  */
 public record TableBufferingFinalizeParams(
         byte[] executionId,
         byte[] finalizeStateId,
+        byte[] attachId,
         BoundStorage storage,
         TableInitParams initParams) {}