ci: multi-transport (subprocess/http/unix) SQL E2E + externalize streaming cursors

.github/workflows/ci.yml

@@ -58,10 +58,23 @@ jobs:
   # built Go worker through the real signed `vgi` community DuckDB extension via
   # a prebuilt standalone `haybarn-unittest` — no C++ build. See ci/README.md.
   # The worker EXECs ffprobe, so ffmpeg is installed here too.
+  #
+  # Transport matrix: the same suite runs over each transport the vgi extension
+  # supports, selected by ci/run-integration.sh's TRANSPORT env var (which
+  # changes what the .test files ATTACH as the worker LOCATION):
+  #   subprocess  worker spawned over stdio (the binary path)
+  #   http        worker started with --http, LOCATION = http://127.0.0.1:<port>
+  #   unix        worker started with --unix <sock>, LOCATION = unix://<sock>
+  # The fixtures are referenced by absolute path, so the out-of-band http/unix
+  # worker reads the same files. See ci/README.md for the per-transport notes.
   integration:
-    name: SQL end-to-end (haybarn)
+    name: SQL E2E (${{ matrix.transport }})
     needs: resolve-haybarn
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        transport: [subprocess, http, unix]
     steps:
       - uses: actions/checkout@v4
 
@@ -96,5 +109,7 @@ jobs:
           echo "HAYBARN_UNITTEST=$UNITTEST" >> "$GITHUB_ENV"
           echo "VGI_MEDIA_WORKER=$PWD/vgi-media-worker" >> "$GITHUB_ENV"
 
-      - name: Run extension integration suite
+      - name: Run extension integration suite (${{ matrix.transport }})
         run: ci/run-integration.sh
+        env:
+          TRANSPORT: ${{ matrix.transport }}

ci/run-integration.sh

@@ -6,21 +6,47 @@
 # community `vgi` extension — no C++ build from source. See ci/README.md.
 #
 # The media worker shells out to `ffprobe` (from ffmpeg) to read committed
-# fixture files. The .test files reference those fixtures by absolute path via
-# VGI_MEDIA_DATA_DIR (mirroring `make test-sql`); ffprobe must be on PATH.
+# fixture files. The .test files reference those fixtures by ABSOLUTE path via
+# VGI_MEDIA_DATA_DIR (mirroring `make test-sql`); ffprobe must be on PATH. No
+# mock server is needed — the committed fixtures are the corpus.
+#
+# Multi-transport: the same suite runs over whichever transport the TRANSPORT
+# env var selects, by changing what `VGI_MEDIA_WORKER` resolves to (the vgi
+# extension picks the transport from the ATTACH LOCATION string):
+#
+#   subprocess (default)  VGI_MEDIA_WORKER = the stdio worker binary
+#                         -> extension spawns it over stdin/stdout.
+#   http                  start `<worker> --http` (prints "PORT:<n>"), parse the
+#                         port, VGI_MEDIA_WORKER = http://127.0.0.1:<port>.
+#   unix                  start `<worker> --unix /tmp/media.sock` (prints
+#                         "UNIX:<path>"), VGI_MEDIA_WORKER = unix:///tmp/media.sock.
+#
+# For http/unix the worker runs out-of-band (not spawned by DuckDB); because the
+# fixtures are referenced by ABSOLUTE path, the out-of-band worker resolves them
+# regardless of its cwd, so ffprobe opens the same files in every transport.
 #
 # Required environment:
 #   HAYBARN_UNITTEST   path to the haybarn-unittest binary
-#   VGI_MEDIA_WORKER   worker LOCATION the .test files ATTACH (the built Go
-#                      worker binary the vgi extension spawns over stdio)
+#   VGI_MEDIA_WORKER   for TRANSPORT=subprocess: the worker LOCATION the .test
+#                      files ATTACH (the built Go worker binary, spawned over
+#                      stdio). For http/unix this is OVERRIDDEN by this script,
+#                      but the binary it points at is reused to launch the
+#                      out-of-band server, so it must still be the worker path.
 # Optional:
+#   TRANSPORT          subprocess (default) | http | unix
 #   VGI_MEDIA_DATA_DIR absolute fixtures dir (default: <repo>/test/sql/data)
 #   STAGE              scratch dir for the preprocessed test tree (default: mktemp)
 set -euo pipefail
 
 : "${HAYBARN_UNITTEST:?path to the haybarn-unittest binary}"
 : "${VGI_MEDIA_WORKER:?worker LOCATION (the built Go worker binary)}"
 
+TRANSPORT="${TRANSPORT:-subprocess}"
+case "$TRANSPORT" in
+  subprocess|http|unix) ;;
+  *) echo "ERROR: unknown TRANSPORT='$TRANSPORT' (expected subprocess|http|unix)" >&2; exit 2 ;;
+esac
+
 HERE="$(cd "$(dirname "$0")" && pwd)"
 REPO="$(cd "$HERE/.." && pwd)"
 STAGE="${STAGE:-$(mktemp -d)}"
@@ -31,19 +57,108 @@ STAGE="${STAGE:-$(mktemp -d)}"
 export VGI_MEDIA_DATA_DIR="${VGI_MEDIA_DATA_DIR:-$REPO/test/sql/data}"
 echo "Using fixtures from $VGI_MEDIA_DATA_DIR"
 
+# The worker binary the subprocess transport ATTACHes to is also the binary we
+# launch out-of-band for http/unix. Capture it before we possibly overwrite
+# VGI_MEDIA_WORKER with a URL.
+WORKER_BIN="$VGI_MEDIA_WORKER"
+
+WORKER_PID=""
+UNIX_SOCK=""
+cleanup() {
+  # Preserve the script's exit status (this runs on EXIT).
+  local rc=$?
+  if [ -n "$WORKER_PID" ]; then kill "$WORKER_PID" 2>/dev/null || true; wait "$WORKER_PID" 2>/dev/null || true; fi
+  if [ -n "$UNIX_SOCK" ]; then rm -f "$UNIX_SOCK"; fi
+  return "$rc"
+}
+trap cleanup EXIT
+
+# --- Per-transport: resolve VGI_MEDIA_WORKER (the ATTACH LOCATION) -----------
+case "$TRANSPORT" in
+  subprocess)
+    echo "Transport: subprocess/stdio — VGI_MEDIA_WORKER=$VGI_MEDIA_WORKER"
+    ;;
+
+  http)
+    WORKER_PORT_FILE="$(mktemp)"
+    echo "Transport: http — starting '$WORKER_BIN --http' ..."
+    "$WORKER_BIN" --http >"$WORKER_PORT_FILE" 2>/dev/null &
+    WORKER_PID=$!
+    WPORT=""
+    for _ in $(seq 1 50); do
+      WPORT="$(sed -n 's/^PORT:\([0-9][0-9]*\)$/\1/p' "$WORKER_PORT_FILE" 2>/dev/null | head -1)"
+      [ -n "$WPORT" ] && break
+      kill -0 "$WORKER_PID" 2>/dev/null || { echo "ERROR: http worker exited before reporting a port" >&2; cat "$WORKER_PORT_FILE" >&2 || true; exit 1; }
+      sleep 0.2
+    done
+    rm -f "$WORKER_PORT_FILE"
+    if [ -z "$WPORT" ]; then
+      echo "ERROR: http worker did not report a port" >&2
+      exit 1
+    fi
+    # Bare scheme://host:port with NO path (the extension POSTs each RPC method
+    # at <LOCATION>/<method>, mounted at the server root).
+    export VGI_MEDIA_WORKER="http://127.0.0.1:$WPORT"
+    echo "HTTP worker listening on $VGI_MEDIA_WORKER (pid $WORKER_PID)"
+    ;;
+
+  unix)
+    UNIX_SOCK="${TMPDIR:-/tmp}/media.$$.sock"
+    rm -f "$UNIX_SOCK"
+    WORKER_OUT_FILE="$(mktemp)"
+    echo "Transport: unix — starting '$WORKER_BIN --unix $UNIX_SOCK' ..."
+    "$WORKER_BIN" --unix "$UNIX_SOCK" >"$WORKER_OUT_FILE" 2>/dev/null &
+    WORKER_PID=$!
+    READY=""
+    for _ in $(seq 1 50); do
+      if grep -q '^UNIX:' "$WORKER_OUT_FILE" 2>/dev/null && [ -S "$UNIX_SOCK" ]; then
+        READY=1; break
+      fi
+      kill -0 "$WORKER_PID" 2>/dev/null || { echo "ERROR: unix worker exited before the socket was ready" >&2; cat "$WORKER_OUT_FILE" >&2 || true; exit 1; }
+      sleep 0.2
+    done
+    rm -f "$WORKER_OUT_FILE"
+    if [ -z "$READY" ]; then
+      echo "ERROR: unix worker did not report a ready socket at $UNIX_SOCK" >&2
+      exit 1
+    fi
+    export VGI_MEDIA_WORKER="unix://$UNIX_SOCK"
+    echo "Unix worker listening on $VGI_MEDIA_WORKER (pid $WORKER_PID)"
+    ;;
+esac
+
 # --- Stage the preprocessed tests -------------------------------------------
 echo "Staging preprocessed tests into $STAGE ..."
 mkdir -p "$STAGE/test/sql"
 for f in "$REPO"/test/sql/*.test; do
   awk -f "$HERE/preprocess-require.awk" "$f" > "$STAGE/test/sql/$(basename "$f")"
 done
 
+# The HTTP transport drives the worker-RPC POSTs through DuckDB's HTTP client,
+# only registered when `httpfs` is loaded. The .test files only `LOAD vgi`, so
+# over HTTP those POSTs fail with an "HTTP"-flavoured error (which the runner
+# silently SKIPS). Inject a signed httpfs INSTALL+LOAD after each `LOAD vgi;`
+# for the http transport only.
+if [ "$TRANSPORT" = "http" ]; then
+  echo "Transport http: injecting 'LOAD httpfs' (required for the worker HTTP RPC) ..."
+  for f in "$STAGE"/test/sql/*.test; do
+    awk '
+      { print }
+      /^LOAD[ \t]+vgi;[ \t]*$/ {
+        print "";
+        print "statement ok";
+        print "INSTALL httpfs FROM core;";
+        print "";
+        print "statement ok";
+        print "LOAD httpfs;";
+      }
+    ' "$f" > "$f.tmp" && mv "$f.tmp" "$f"
+  done
+fi
+
 cd "$STAGE"
 
-# Warm the extension cache once: vgi from the signed community channel. A miss
-# here is only a warning — the per-test LOAD vgi; (the .test files load it
-# explicitly) is what actually gates each file, and it needs vgi already
-# INSTALLed into the runner's extension dir.
+# Warm the extension cache once: vgi from the signed community channel.
 echo "Warming the extension cache (vgi from community) ..."
 mkdir -p "$STAGE/test"
 cat > "$STAGE/test/_warm.test" <<'EOF'
@@ -55,7 +170,29 @@ EOF
 "$HAYBARN_UNITTEST" "test/_warm.test" >/dev/null 2>&1 || echo "::warning::extension warm step did not fully succeed"
 rm -f "$STAGE/test/_warm.test"
 
-# Run the whole suite in one invocation, streaming the runner's native
-# sqllogictest report. Any failed assertion exits non-zero and fails the job.
-echo "Running suite (worker: $VGI_MEDIA_WORKER) ..."
-"$HAYBARN_UNITTEST" "test/sql/*"
+# Run the whole suite in one invocation, capturing the runner's native
+# sqllogictest report so we can both stream it AND guard against a silent skip.
+#
+# IMPORTANT: the runner SKIPS (exit 0) a test whose error message matches a
+# built-in network-error allowlist that includes "HTTP". A broken HTTP transport
+# would otherwise show "All tests were skipped" and go GREEN having run nothing.
+# We detect that and fail explicitly.
+echo "Running suite (transport: $TRANSPORT, worker: $VGI_MEDIA_WORKER) ..."
+RUN_LOG="$STAGE/run.log"
+set +e
+"$HAYBARN_UNITTEST" "test/sql/*" 2>&1 | tee "$RUN_LOG"
+RUN_RC="${PIPESTATUS[0]}"
+set -e
+
+if [ "$RUN_RC" -ne 0 ]; then
+  echo "ERROR: suite failed (transport: $TRANSPORT, rc=$RUN_RC)" >&2
+  exit "$RUN_RC"
+fi
+
+if grep -q 'All tests were skipped' "$RUN_LOG"; then
+  echo "ERROR: every test was SKIPPED on transport '$TRANSPORT' (the runner's" >&2
+  echo "       built-in network-error skip swallowed the real error). This is" >&2
+  echo "       NOT a pass. Skip reason reported by the runner:" >&2
+  grep -A3 'Skipped tests for the following reasons' "$RUN_LOG" >&2 || true
+  exit 1
+fi

cmd/vgi-media-worker/main.go

@@ -21,15 +21,18 @@ import (
 )
 
 func main() {
-	// Accept --http for HTTP transport; default is stdio. Unknown launcher flags
-	// are tolerated (the VGI extension varies argv to key its worker cache), so
-	// we filter to flags we actually define before parsing.
+	// Accept --http for HTTP transport and --unix for the AF_UNIX launcher
+	// transport; default is stdio. Unknown launcher flags are tolerated (the
+	// VGI extension varies argv to key its worker cache), so we filter to flags
+	// we actually define before parsing.
 	httpMode := flag.Bool("http", false, "Run as an HTTP server instead of stdio")
+	unixPath := flag.String("unix", "", "Serve the AF_UNIX launcher transport on this socket path instead of stdio")
 	logFlags := vgi.RegisterLoggingFlags(flag.CommandLine)
 	_ = flag.CommandLine.Parse(filterKnownFlags(os.Args[1:], map[string]bool{
 		"log-level":  true,
 		"log-format": true,
 		"log-logger": true,
+		"unix":       true,
 	}))
 	if err := logFlags.Apply(); err != nil {
 		log.Fatalf("logging flags: %v", err)
@@ -50,6 +53,15 @@ func main() {
 		}
 		return
 	}
+	if *unixPath != "" {
+		// AF_UNIX launcher transport: serve on the given socket path. The SDK
+		// prints "UNIX:<path>" once listening; idleTimeout=0 disables the
+		// self-shutdown timer (the launcher/CI owns the process lifecycle).
+		if err := w.RunUnix(*unixPath, 0); err != nil {
+			log.Fatal(err)
+		}
+		return
+	}
 	w.RunStdio()
 }
 

internal/mediaworker/functions_test.go

@@ -3,7 +3,9 @@
 package mediaworker
 
 import (
+	"bytes"
 	"context"
+	"encoding/gob"
 	"testing"
 
 	"github.com/Query-farm/vgi-go/vgi"
@@ -46,8 +48,8 @@ func TestStreamsNewStateMP4(t *testing.T) {
 	if !r.HasWidth || r.Width != 320 || r.Height != 240 {
 		t.Errorf("dims = %dx%d (hasW=%v)", r.Width, r.Height, r.HasWidth)
 	}
-	if st.Done {
-		t.Error("state should not be Done before Process")
+	if st.Offset != 0 {
+		t.Error("state cursor should start at offset 0 before Process")
 	}
 }
 
@@ -140,3 +142,33 @@ func TestProbeRowGarbageBytes(t *testing.T) {
 		t.Errorf("garbage bytes should yield nil result, got %+v", r)
 	}
 }
+
+// TestCursorSurvivesContinuation mirrors the HTTP transport: the per-scan state
+// is gob round-tripped between ticks, so the cursor offset must advance across
+// the boundary and eventually drain. A bare Done flag flipped after Emit would
+// re-emit row 0 forever; the explicit Offset terminates.
+func TestCursorSurvivesContinuation(t *testing.T) {
+	n := rowsPerTick*2 + 5 // spans 3 ticks
+	st := &streamsState{Rows: make([]streamRow, n)}
+	emitted := 0
+	for tick := 0; tick < 100; tick++ {
+		var buf bytes.Buffer
+		if err := gob.NewEncoder(&buf).Encode(st); err != nil {
+			t.Fatalf("gob encode: %v", err)
+		}
+		var resumed streamsState
+		if err := gob.NewDecoder(&buf).Decode(&resumed); err != nil {
+			t.Fatalf("gob decode: %v", err)
+		}
+		st = &resumed
+		start, end, done := cursorBounds(len(st.Rows), &st.Offset)
+		if done {
+			if emitted != n {
+				t.Fatalf("drained after emitting %d of %d rows", emitted, n)
+			}
+			return
+		}
+		emitted += end - start
+	}
+	t.Fatal("cursor never drained — continuation loop did not terminate")
+}