diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 978f582..5ea264d 100644 --- a/.github/workflows/ci.yml +++ b/.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: + # unix worker started with --unix , LOCATION = unix:// + # 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 }} diff --git a/ci/run-integration.sh b/ci/run-integration.sh index 3f52664..b8201ca 100755 --- a/ci/run-integration.sh +++ b/ci/run-integration.sh @@ -6,14 +6,34 @@ # 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 ` --http` (prints "PORT:"), parse the +# port, VGI_MEDIA_WORKER = http://127.0.0.1:. +# unix start ` --unix /tmp/media.sock` (prints +# "UNIX:"), 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: /test/sql/data) # STAGE scratch dir for the preprocessed test tree (default: mktemp) set -euo pipefail @@ -21,6 +41,12 @@ 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,6 +57,76 @@ 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 /, 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" @@ -38,12 +134,31 @@ 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 diff --git a/cmd/vgi-media-worker/main.go b/cmd/vgi-media-worker/main.go index 16abe18..d2a1b82 100644 --- a/cmd/vgi-media-worker/main.go +++ b/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:" 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() } diff --git a/internal/mediaworker/functions_test.go b/internal/mediaworker/functions_test.go index 9d37b06..3baf60d 100644 --- a/internal/mediaworker/functions_test.go +++ b/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") +} diff --git a/internal/mediaworker/tables.go b/internal/mediaworker/tables.go index b82843f..883b893 100644 --- a/internal/mediaworker/tables.go +++ b/internal/mediaworker/tables.go @@ -18,6 +18,42 @@ var allocator = memory.NewGoAllocator() func itoa(n int) string { return strconv.Itoa(n) } +// WHY AN EXPLICIT CURSOR, NOT A bool Done (the HTTP-continuation fix): +// +// Over the HTTP transport the worker is STATELESS across exchanges — there is no +// long-lived process holding the live state between Process ticks. The framework +// round-trips the producer state through an opaque continuation token: after each +// tick it gob-encodes the state (snapshotting the LIVE user state), the client +// returns the token, and the worker resumes by gob-decoding it. The HTTP server +// emits at most one data batch per response, so a producer with more to emit is +// always resumed mid-stream from its token. +// +// The position MUST therefore live in the serialized state. A bare `Done bool` +// flipped only AFTER the single Emit does not survive the continuation boundary: +// the resumed tick observes the pre-Emit snapshot, re-emits the same rows, and +// the scan never terminates (an infinite loop — subprocess/unix keep live state +// in memory, so they were unaffected and hid the bug). Carrying an explicit +// Offset that Process advances BEFORE yielding makes the snapshot authoritative. +// +// rowsPerTick bounds how many rows each Process tick emits, so the cursor is +// observable across the continuation boundary (and scales to large results). +const rowsPerTick = 256 + +// cursorBounds returns [start,end) for the next bounded slice over n rows +// starting at *offset, advancing *offset past it; done=true once all consumed. +func cursorBounds(n int, offset *int) (start, end int, done bool) { + if *offset >= n { + return 0, 0, true + } + start = *offset + end = start + rowsPerTick + if end > n { + end = n + } + *offset = end + return start, end, false +} + // tableArgs is the single-argument struct for the table functions: a path // (VARCHAR) or media bytes (BLOB). NOTE: table functions cannot take a column // arg (they have no streamed input batch), so the input is a CONST scalar @@ -83,8 +119,8 @@ type streamRow struct { } type streamsState struct { - Done bool - Rows []streamRow + Rows []streamRow + Offset int } // StreamsFunction lists every elementary stream in the input media. @@ -142,11 +178,11 @@ func (f *StreamsFunction) NewState(params *vgi.ProcessParams) (*streamsState, er return &streamsState{Rows: rows}, nil } func (f *StreamsFunction) Process(_ context.Context, _ *vgi.ProcessParams, state *streamsState, out *vgirpc.OutputCollector) error { - if state.Done { + start, end, done := cursorBounds(len(state.Rows), &state.Offset) + if done { return out.Finish() } - state.Done = true - rows := state.Rows + rows := state.Rows[start:end] n := len(rows) idx := array.NewInt32Builder(allocator) @@ -210,8 +246,8 @@ type tagKV struct { } type tagsState struct { - Done bool - Tags []tagKV + Tags []tagKV + Offset int } // TagsFunction lists the container-level (format) metadata tags. @@ -252,11 +288,11 @@ func (f *TagsFunction) NewState(params *vgi.ProcessParams) (*tagsState, error) { return &tagsState{Tags: tags}, nil } func (f *TagsFunction) Process(_ context.Context, _ *vgi.ProcessParams, state *tagsState, out *vgirpc.OutputCollector) error { - if state.Done { + start, end, done := cursorBounds(len(state.Tags), &state.Offset) + if done { return out.Finish() } - state.Done = true - t := state.Tags + t := state.Tags[start:end] n := int64(len(t)) batch := array.NewRecordBatch(tagsSchema, []arrow.Array{ vgi.BuildStringArray(n, func(i int64) string { return t[i].Key }),