diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 32fec60..5442268 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,10 +53,23 @@ jobs: # SQL end-to-end: run the sqllogictest suite (test/sql/*.test) against the # built Go worker through the real signed `vgi` community DuckDB extension via # a prebuilt standalone `haybarn-unittest` — no C++ build. See ci/README.md. + # + # 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 mock OData server is started for ALL transports (the worker's table + # functions still call it). 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 @@ -88,5 +101,7 @@ jobs: echo "HAYBARN_UNITTEST=$UNITTEST" >> "$GITHUB_ENV" echo "VGI_ODATA_WORKER=$PWD/vgi-odata-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 2934132..f7c4f47 100755 --- a/ci/run-integration.sh +++ b/ci/run-integration.sh @@ -5,29 +5,75 @@ # VGI worker, using a prebuilt standalone `haybarn-unittest` and the signed # community `vgi` extension — no C++ build from source. See ci/README.md. # -# The odata worker queries OData services over HTTP, so the suite needs a -# server: this script builds the repo's `mockserver`, starts it on a free port, -# and points the tests at it via VGI_ODATA_TEST_URL (mirroring `make test-sql`). +# Multi-transport: the same suite runs over whichever transport the TRANSPORT +# env var selects, by changing what `VGI_ODATA_WORKER` resolves to (the vgi +# extension picks the transport from the ATTACH LOCATION string): +# +# subprocess (default) VGI_ODATA_WORKER = the stdio worker binary +# -> extension spawns it over stdin/stdout. +# http start ` --http` (prints "PORT:"), parse the +# port, VGI_ODATA_WORKER = http://127.0.0.1:. +# (The extension POSTs each RPC method at /; +# the SDK mounts them at the root, so LOCATION has no path.) +# unix start ` --unix /tmp/odata.sock` (prints +# "UNIX:"), VGI_ODATA_WORKER = unix:///tmp/odata.sock. +# +# In every transport the odata worker queries OData over HTTP, so the suite +# ALWAYS needs the mock OData server: this script builds the repo's `mockserver`, +# starts it on a free port, and points the tests at it via VGI_ODATA_TEST_URL +# (mirroring `make test-sql`). All started processes are trap-killed on exit. # # Required environment: # HAYBARN_UNITTEST path to the haybarn-unittest binary -# VGI_ODATA_WORKER worker LOCATION the .test files ATTACH (the built Go -# worker binary the vgi extension spawns over stdio) +# VGI_ODATA_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 # STAGE scratch dir for the preprocessed test tree (default: mktemp) set -euo pipefail : "${HAYBARN_UNITTEST:?path to the haybarn-unittest binary}" : "${VGI_ODATA_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)}" -# --- Start the mock OData server (the .test files query it) ----------------- +# 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_ODATA_WORKER with a URL. +WORKER_BIN="$VGI_ODATA_WORKER" + +# Collected PIDs and paths to clean up on exit (mock + optional worker server). +MOCK_PID="" +WORKER_PID="" +UNIX_SOCK="" +cleanup() { + # Preserve the script's exit status: this runs on EXIT, so its own last + # command must not clobber the real exit code (a bare `[ -n "$x" ]` that is + # false returns 1 and would turn a green run red). + local rc=$? + if [ -n "$WORKER_PID" ]; then kill "$WORKER_PID" 2>/dev/null || true; wait "$WORKER_PID" 2>/dev/null || true; fi + if [ -n "$MOCK_PID" ]; then kill "$MOCK_PID" 2>/dev/null || true; wait "$MOCK_PID" 2>/dev/null || true; fi + if [ -n "$UNIX_SOCK" ]; then rm -f "$UNIX_SOCK"; fi + return "$rc" +} +trap cleanup EXIT + +# --- Start the mock OData server (the .test files query it; all transports) --- # Build + launch the repo's standalone mock server on a free port; it prints -# "PORT:" on stdout (see cmd/mockserver/main.go). We capture that, export -# VGI_ODATA_TEST_URL, and kill the server on exit — exactly like `make test-sql`. +# "PORT:" on stdout (see cmd/mockserver/main.go). We capture that and export +# VGI_ODATA_TEST_URL. The mock is required for every transport — the worker still +# makes the HTTP call. MOCK_BIN="$STAGE/mockserver" echo "Building mock OData server ..." ( cd "$REPO" && go build -o "$MOCK_BIN" ./cmd/mockserver ) @@ -35,12 +81,6 @@ echo "Building mock OData server ..." MOCK_PORT_FILE="$(mktemp)" "$MOCK_BIN" --addr 127.0.0.1:0 >"$MOCK_PORT_FILE" 2>/dev/null & MOCK_PID=$! -cleanup() { - kill "$MOCK_PID" 2>/dev/null || true - wait "$MOCK_PID" 2>/dev/null || true - rm -f "$MOCK_PORT_FILE" -} -trap cleanup EXIT PORT="" for _ in $(seq 1 30); do @@ -52,9 +92,69 @@ if [ -z "$PORT" ]; then echo "ERROR: mock server did not report a port" >&2 exit 1 fi +rm -f "$MOCK_PORT_FILE" export VGI_ODATA_TEST_URL="http://127.0.0.1:$PORT" echo "Mock OData server listening on $VGI_ODATA_TEST_URL (pid $MOCK_PID)" +# --- Per-transport: resolve VGI_ODATA_WORKER (the ATTACH LOCATION) ----------- +# subprocess keeps the binary path (extension spawns stdio). http/unix start the +# worker out-of-band and hand the extension a URL. +case "$TRANSPORT" in + subprocess) + echo "Transport: subprocess/stdio — VGI_ODATA_WORKER=$VGI_ODATA_WORKER" + ;; + + http) + # Start the worker in --http mode; it prints "PORT:" once listening. + 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 + # The LOCATION must be the bare scheme://host:port with NO path (the + # extension POSTs each RPC method at /, mounted at root). + export VGI_ODATA_WORKER="http://127.0.0.1:$WPORT" + echo "HTTP worker listening on $VGI_ODATA_WORKER (pid $WORKER_PID)" + ;; + + unix) + # Start the worker on an AF_UNIX socket; it prints "UNIX:" once + # listening. idleTimeout is disabled (we own the process lifecycle). + UNIX_SOCK="${TMPDIR:-/tmp}/odata.$$.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_ODATA_WORKER="unix://$UNIX_SOCK" + echo "Unix worker listening on $VGI_ODATA_WORKER (pid $WORKER_PID)" + ;; +esac + # --- Stage the preprocessed tests ------------------------------------------- echo "Staging preprocessed tests into $STAGE ..." mkdir -p "$STAGE/test/sql" @@ -62,6 +162,29 @@ 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, +# which is only registered when the `httpfs` extension is loaded. The .test +# files only `LOAD vgi`, so over HTTP those POSTs fail with an "HTTP"-flavoured +# error (which the runner then silently SKIPS). Inject a signed +# `INSTALL httpfs FROM core; LOAD httpfs;` after each `LOAD vgi;` for the http +# transport only (subprocess/unix do not use the HTTP client). +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 @@ -79,7 +202,30 @@ 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_ODATA_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 DuckDB/Haybarn sqllogictest runner SKIPS (not fails, exit 0) a +# test whose error message matches a built-in network-error allowlist that +# includes the substring "HTTP". So a broken HTTP transport would otherwise show +# "All tests were skipped" and the job would go GREEN having run nothing — a +# fake pass. We detect that and fail explicitly. +echo "Running suite (transport: $TRANSPORT, worker: $VGI_ODATA_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-odata-worker/main.go b/cmd/vgi-odata-worker/main.go index ef618b3..0c9b236 100644 --- a/cmd/vgi-odata-worker/main.go +++ b/cmd/vgi-odata-worker/main.go @@ -16,15 +16,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) @@ -45,6 +48,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/odataworker/functions.go b/internal/odataworker/functions.go index c4917f8..a9ca264 100644 --- a/internal/odataworker/functions.go +++ b/internal/odataworker/functions.go @@ -16,9 +16,42 @@ import ( // EXPORTED, gob-encodable fields only — no arrow.Record, no interfaces/chans/ // funcs, no unexported fields. Each function fetches its rows eagerly in // NewState, stores them as plain Go slices, and rebuilds the Arrow batch in -// Process. The Done field is exported so gob can round-trip it. -type emitState struct { - Done bool +// Process. +// +// 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. Instead 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 that has +// 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 + +// cursorSlice returns the next bounded slice of rows starting at *offset and +// advances *offset past them, reporting done=true once all rows are consumed. +func cursorSlice[T any](rows []T, offset *int) (slice []T, done bool) { + if *offset >= len(rows) { + return nil, true + } + end := *offset + rowsPerTick + if end > len(rows) { + end = len(rows) + } + slice = rows[*offset:end] + *offset = end + return slice, false } // optsFrom assembles QueryOptions from the bound arguments. @@ -55,10 +88,10 @@ type queryArgs struct { Version string `vgi:"name=version,default=v4,doc=OData response shape: 'v4' (value/@odata.nextLink) or 'v2' (d.results/d.__next)"` } -// queryState holds the fetched entities (gob-encodable) plus the emit flag. +// queryState holds the fetched entities (gob-encodable) plus the cursor offset. type queryState struct { - emitState Entities []Entity + Offset int } // QueryFunction reads an OData entity set as rows of raw JSON. @@ -99,11 +132,10 @@ func (f *QueryFunction) NewState(params *vgi.ProcessParams) (*queryState, error) } func (f *QueryFunction) Process(_ context.Context, _ *vgi.ProcessParams, state *queryState, out *vgirpc.OutputCollector) error { - if state.Done { + e, done := cursorSlice(state.Entities, &state.Offset) + if done { return out.Finish() } - state.Done = true - e := state.Entities n := int64(len(e)) batch := array.NewRecordBatch(querySchema, []arrow.Array{ vgi.BuildInt64Array(n, func(i int64) int64 { return e[i].Seq }), @@ -131,10 +163,10 @@ type entitySetsArgs struct { Token string `vgi:"name=token,default=,doc=Bearer token (Authorization: Bearer )"` } -// entitySetsState holds the discovered entity-set names plus the emit flag. +// entitySetsState holds the discovered entity-set names plus the cursor offset. type entitySetsState struct { - emitState - Names []string + Names []string + Offset int } // EntitySetsFunction lists the entity sets of a service from its service doc. @@ -176,12 +208,12 @@ func (f *EntitySetsFunction) NewState(params *vgi.ProcessParams) (*entitySetsSta } func (f *EntitySetsFunction) Process(_ context.Context, _ *vgi.ProcessParams, state *entitySetsState, out *vgirpc.OutputCollector) error { - if state.Done { + names, done := cursorSlice(state.Names, &state.Offset) + if done { return out.Finish() } - state.Done = true - n := int64(len(state.Names)) - col := vgi.BuildStringArray(n, func(i int64) string { return state.Names[i] }) + n := int64(len(names)) + col := vgi.BuildStringArray(n, func(i int64) string { return names[i] }) batch := array.NewRecordBatch(entitySetsSchema, []arrow.Array{col}, n) defer batch.Release() return out.Emit(batch) @@ -207,10 +239,10 @@ type metadataArgs struct { Token string `vgi:"name=token,default=,doc=Bearer token (Authorization: Bearer )"` } -// metadataState holds the parsed property rows plus the emit flag. +// metadataState holds the parsed property rows plus the cursor offset. type metadataState struct { - emitState - Rows []PropertyRow + Rows []PropertyRow + Offset int } // MetadataFunction parses $metadata (EDMX) into property rows. @@ -252,11 +284,10 @@ func (f *MetadataFunction) NewState(params *vgi.ProcessParams) (*metadataState, } func (f *MetadataFunction) Process(_ context.Context, _ *vgi.ProcessParams, state *metadataState, out *vgirpc.OutputCollector) error { - if state.Done { + r, done := cursorSlice(state.Rows, &state.Offset) + if done { return out.Finish() } - state.Done = true - r := state.Rows n := int64(len(r)) batch := array.NewRecordBatch(metadataSchema, []arrow.Array{ vgi.BuildStringArray(n, func(i int64) string { return r[i].EntityType }), diff --git a/internal/odataworker/functions_test.go b/internal/odataworker/functions_test.go index bcca399..ed5f384 100644 --- a/internal/odataworker/functions_test.go +++ b/internal/odataworker/functions_test.go @@ -3,6 +3,8 @@ package odataworker import ( + "bytes" + "encoding/gob" "testing" "github.com/Query-farm/vgi-go/vgi" @@ -45,8 +47,8 @@ func TestQueryNewStateData(t *testing.T) { if len(st.Entities) == 0 { t.Fatal("expected entities from NewState") } - if st.Done { - t.Error("state should not be marked done before Process") + if st.Offset != 0 { + t.Error("state cursor should start at offset 0 before Process") } } @@ -91,6 +93,38 @@ func TestMetadataNewStateData(t *testing.T) { } } +// 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 +// not survive this (it would re-emit row 0 forever); an explicit Offset does. +func TestCursorSurvivesContinuation(t *testing.T) { + rows := make([]Entity, rowsPerTick*2+5) // spans 3 ticks + st := &queryState{Entities: rows} + emitted := 0 + for tick := 0; tick < 100; tick++ { + // Snapshot the live state through gob (as the HTTP framework does), then + // resume from the decoded copy. + var buf bytes.Buffer + if err := gob.NewEncoder(&buf).Encode(st); err != nil { + t.Fatalf("gob encode: %v", err) + } + var resumed queryState + if err := gob.NewDecoder(&buf).Decode(&resumed); err != nil { + t.Fatalf("gob decode: %v", err) + } + st = &resumed + slice, done := cursorSlice(st.Entities, &st.Offset) + if done { + if emitted != len(rows) { + t.Fatalf("drained after emitting %d of %d rows", emitted, len(rows)) + } + return + } + emitted += len(slice) + } + t.Fatal("cursor never drained — continuation loop did not terminate") +} + func TestRegisterDoesNotPanic(t *testing.T) { // Registration runs the SDK's gob-encodability validation on each state type; // a non-encodable state field would panic here.