diff --git a/.github/workflows/backend-test-windows.yml b/.github/workflows/backend-test-windows.yml index 3468449753..9fa13bca58 100644 --- a/.github/workflows/backend-test-windows.yml +++ b/.github/workflows/backend-test-windows.yml @@ -139,8 +139,8 @@ jobs: run: | cargo build --release -p windmill_duckdb_ffi_internal New-Item -ItemType Directory -Path ..\target\debug -Force - Copy-Item target\release\windmill_duckdb_ffi_internal.dll ..\target\debug\ - # duckdb is bundled (~2GB of build artifacts); the DLL is the only + Copy-Item target\release\windmill_duckdb_runner.exe ..\target\debug\ + # duckdb is bundled (~2GB of build artifacts); the runner is the only # thing we need from this excluded-crate target dir. Remove-Item -Recurse -Force -ErrorAction SilentlyContinue target Get-PSDrive C | Select-Object Used,Free | Format-Table -AutoSize diff --git a/.github/workflows/build-publish-rh-image.yml b/.github/workflows/build-publish-rh-image.yml index dcb223b534..f037e7eb75 100644 --- a/.github/workflows/build-publish-rh-image.yml +++ b/.github/workflows/build-publish-rh-image.yml @@ -83,17 +83,17 @@ jobs: # Extract arm64 binary (include deps/ for hard link resolution) mkdir -p /tmp/arm64 crane export --platform linux/arm64 ${{ steps.meta-ee-public.outputs.tags }} - \ - | tar -xf - -C /tmp/arm64 windmill/target/release/ usr/src/app/libwindmill_duckdb_ffi_internal.so + | tar -xf - -C /tmp/arm64 windmill/target/release/ usr/src/app/windmill_duckdb_runner cp /tmp/arm64/windmill/target/release/windmill extracted/windmill-ee-arm64-rhel9 - cp /tmp/arm64/usr/src/app/libwindmill_duckdb_ffi_internal.so extracted/libwindmill_duckdb_ffi_internal-arm64.so + cp /tmp/arm64/usr/src/app/windmill_duckdb_runner extracted/windmill_duckdb_runner-arm64 rm -rf /tmp/arm64 # Extract amd64 binary mkdir -p /tmp/amd64 crane export --platform linux/amd64 ${{ steps.meta-ee-public.outputs.tags }} - \ - | tar -xf - -C /tmp/amd64 windmill/target/release/ usr/src/app/libwindmill_duckdb_ffi_internal.so + | tar -xf - -C /tmp/amd64 windmill/target/release/ usr/src/app/windmill_duckdb_runner cp /tmp/amd64/windmill/target/release/windmill extracted/windmill-ee-amd64-rhel9 - cp /tmp/amd64/usr/src/app/libwindmill_duckdb_ffi_internal.so extracted/libwindmill_duckdb_ffi_internal-amd64.so + cp /tmp/amd64/usr/src/app/windmill_duckdb_runner extracted/windmill_duckdb_runner-amd64 rm -rf /tmp/amd64 - uses: actions/upload-artifact@v4 @@ -109,9 +109,9 @@ jobs: - uses: actions/upload-artifact@v4 with: name: RHEL9-arm64 dynamic libraries build - path: extracted/libwindmill_duckdb_ffi_internal-arm64.so + path: extracted/windmill_duckdb_runner-arm64 - uses: actions/upload-artifact@v4 with: name: RHEL9-amd64 dynamic libraries build - path: extracted/libwindmill_duckdb_ffi_internal-amd64.so + path: extracted/windmill_duckdb_runner-amd64 diff --git a/.github/workflows/build-publish-rh8-image.yml b/.github/workflows/build-publish-rh8-image.yml index 7c820eb290..83b12d3290 100644 --- a/.github/workflows/build-publish-rh8-image.yml +++ b/.github/workflows/build-publish-rh8-image.yml @@ -103,7 +103,7 @@ jobs: id: extract-duckdb-ffi-internal with: image: ${{ steps.meta-ee-public.outputs.tags}}-amd64 - path: "/usr/src/app/libwindmill_duckdb_ffi_internal.so" + path: "/usr/src/app/windmill_duckdb_runner" # - uses: shrink/actions-docker-extract@v3 # id: extract-ee-arm64 @@ -124,7 +124,7 @@ jobs: - uses: actions/upload-artifact@v4 with: name: RHEL8-amd64 dynamic libraries build - path: ${{ steps.extract-duckdb-ffi-internal.outputs.destination }}/libwindmill_duckdb_ffi_internal.so + path: ${{ steps.extract-duckdb-ffi-internal.outputs.destination }}/windmill_duckdb_runner # - uses: actions/upload-artifact@v4 # with: diff --git a/.github/workflows/build_windows_worker_.yml b/.github/workflows/build_windows_worker_.yml index 46befe42b6..4a6368ee1e 100644 --- a/.github/workflows/build_windows_worker_.yml +++ b/.github/workflows/build_windows_worker_.yml @@ -81,5 +81,5 @@ jobs: - name: Upload dynamic libraries artifact uses: actions/upload-artifact@v4 with: - name: windmill_duckdb_ffi_internal.dll - path: ./backend/windmill-duckdb-ffi-internal/target/release/windmill_duckdb_ffi_internal.dll + name: windmill_duckdb_runner.exe + path: ./backend/windmill-duckdb-ffi-internal/target/release/windmill_duckdb_runner.exe diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index c94521d332..7e05675495 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -218,7 +218,7 @@ jobs: id: extract-duckdb-ffi-internal with: image: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}@${{ needs.build.outputs.digest }} - path: "/usr/src/app/libwindmill_duckdb_ffi_internal.so" + path: "/usr/src/app/windmill_duckdb_runner" - uses: shrink/actions-docker-extract@v3 id: extract-ee diff --git a/.github/workflows/publish_windows_worker.yml b/.github/workflows/publish_windows_worker.yml index b94d5d57c7..ecceff03ab 100644 --- a/.github/workflows/publish_windows_worker.yml +++ b/.github/workflows/publish_windows_worker.yml @@ -72,8 +72,8 @@ jobs: files: | ./backend/target/release/windmill-ee.exe - - name: Attach dynamic libraries to release + - name: Attach the DuckDB runner to release uses: softprops/action-gh-release@v2 with: files: | - ./backend/windmill-duckdb-ffi-internal/target/release/windmill_duckdb_ffi_internal.dll + ./backend/windmill-duckdb-ffi-internal/target/release/windmill_duckdb_runner.exe diff --git a/Dockerfile b/Dockerfile index 4e022d4b53..80de6d4d16 100644 --- a/Dockerfile +++ b/Dockerfile @@ -50,10 +50,7 @@ RUN apt-get update && apt-get install -y clang=1:19.0* libclang-dev=1:19.0* cmak COPY ./backend/windmill-duckdb-ffi-internal . -# The `duckdb` crate comes from a git dependency (a fork carrying an engine patch), -# which cargo checks out under $CARGO_HOME/git rather than the registry cache. RUN --mount=type=cache,target=/usr/local/cargo/registry \ - --mount=type=cache,target=/usr/local/cargo/git \ --mount=type=cache,target=$SCCACHE_DIR,sharing=locked \ cargo build --release -p windmill_duckdb_ffi_internal @@ -286,7 +283,8 @@ ENV TZ=Etc/UTC COPY --from=builder /frontend/build /static_frontend COPY --from=builder /windmill/target/release/windmill ${APP}/windmill -COPY --from=windmill_duckdb_ffi_internal_builder /windmill-duckdb-ffi-internal/target/release/libwindmill_duckdb_ffi_internal.so ${APP}/libwindmill_duckdb_ffi_internal.so +# DuckDB jobs run in this binary, which the worker looks for next to its own. +COPY --from=windmill_duckdb_ffi_internal_builder /windmill-duckdb-ffi-internal/target/release/windmill_duckdb_runner ${APP}/windmill_duckdb_runner COPY --from=oven/bun:1.4.0 /usr/local/bin/bun /usr/bin/bun diff --git a/backend/.gitignore b/backend/.gitignore index 6645e22618..78639e3b4e 100644 --- a/backend/.gitignore +++ b/backend/.gitignore @@ -10,4 +10,4 @@ windmill-api/openapi-*.* *ee.rs generate_mcp_endpoints_tools/venv bacon.toml -libwindmill_duckdb_ffi_internal.so \ No newline at end of file +windmill_duckdb_runner \ No newline at end of file diff --git a/backend/AGENTS.md b/backend/AGENTS.md index fbacb857ee..290890009a 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -6,17 +6,17 @@ - **DB schema**: `backend/summarized_schema.txt` - **API routes entry point**: `windmill-api/src/lib.rs` - **OpenAPI spec**: `windmill-api/openapi.yaml` -- **DuckDB local jobs**: build the dynamic FFI library before running DuckDB scripts locally: +- **DuckDB local jobs**: build the DuckDB runner before running DuckDB scripts locally: ```bash cd backend/windmill-duckdb-ffi-internal && ./build_dev.sh ``` - Re-run after clean builds or when `target/debug/libwindmill_duckdb_ffi_internal.*` is missing. + Re-run after clean builds or when `target/debug/windmill_duckdb_runner` is missing. The bundled DuckDB compile (~2min) is cached in a per-user dir shared across worktrees (keyed by the crate's `Cargo.lock`), so a fresh worktree reuses it and the build is near-instant — you don't pay the full compile per worktree. Editing - the FFI crate's own source falls back to an isolated per-worktree `./target`. - The engine is a **patched fork** of duckdb-rs, not the crates.io crate — read - `docs/duckdb-isolation.md` before bumping it or touching the isolation transform. + the crate's own source falls back to an isolated per-worktree `./target`. + Jobs run in that binary, out of the worker's process, under the worker's job isolation — + read `docs/duckdb-execution.md` before changing how it is spawned or what crosses to it. - **Running data pipelines (DuckLake) from source**: see the section below — a plain build advertises the `duckdb` tag but cannot execute DuckDB scripts and has no working S3 proxy. @@ -120,7 +120,7 @@ only what you need — build time scales with the set. | `enterprise` | EE business logic (autoscaling, SAML hooks, advanced S3 rule **enforcement**, WAP, forks, …). Pulls in `license`. | Running EE features. Advanced S3 permission rules only take effect with this. | | `license` | License-key/plan plumbing (`LICENSE_KEY`). Pulled in by `enterprise`. Having the feature compiled does **not** require a license *key* at runtime — CE defaults to a free plan and most EE paths still run keyless. | License-gated behavior. | | `parquet` | S3/object-storage support: the `job_helpers/*` and `apps_u/*` S3 endpoints, parquet/CSV preview, workspace large-file storage. Without it those routes return `"requires parquet"`. | Anything touching S3/object storage or datasets. | -| `duckdb` | DuckDB script executor (also needs the FFI dylib — see above). | DuckDB scripts, DuckLake. | +| `duckdb` | DuckDB script executor (also needs the runner binary — see above). | DuckDB scripts, DuckLake. | | `python` `rust` `php` `java` `ruby` `csharp` `nu` `deno_core` `mysql` `mssql` `bigquery` `snowflake` `oracledb` `rlang` | Each enables that language/DB runtime for job execution. | Running jobs in that language. | | `mcp` | MCP gateway routes (baseline `quickjs` does NOT include it → MCP routes 404). | MCP work. | | `websocket` `http_trigger` `kafka` `nats` `mqtt_trigger` `sqs_trigger` `gcp_trigger` `azure_trigger` `postgres_trigger` `native_trigger` | Each native trigger kind; none on by default (creating one 404s without its feature). | Working on / exercising that trigger. | @@ -136,7 +136,7 @@ minimal explicit set for dev. | Plain dev baseline (JS eval only) | `quickjs` | | S3 / object storage / datasets (CE) | `quickjs,private,parquet` | | S3 + EE (advanced S3 rules, on-behalf app reads, WAP, forks) | `quickjs,enterprise,private,parquet` | -| DuckLake / DuckDB (CE) | `quickjs,duckdb,parquet,private` (+ build the FFI) | +| DuckLake / DuckDB (CE) | `quickjs,duckdb,parquet,private` (+ build the runner) | | + Python jobs | append `,python` | ## Workspace object storage in dev — use the local filesystem @@ -163,7 +163,7 @@ in DuckDB"`) — DuckLake/datatable go through the S3 proxy instead, which works ## Running data pipelines (DuckLake) from source -DuckLake pipelines need **both** the right cargo features **and** the prebuilt DuckDB FFI. A +DuckLake pipelines need **both** the right cargo features **and** the DuckDB runner binary. A plain `cargo run` (or `cargo run --features quickjs`) does **not** suffice, and the failure modes are silent-ish, so agents lose time. Verify feature names against `backend/Cargo.toml` `[features]`. @@ -178,7 +178,7 @@ are silent-ish, so agents lose time. Verify feature names against `backend/Cargo `enterprise` already pulls in `license`, but list both when you want the license-gated paths. `quickjs` is for JS eval, not DuckLake per se — keep it if your baseline build had it. -**Before running any DuckDB script**, build the FFI (see the bullet above): +**Before running any DuckDB script**, build the runner (see the bullet above): `cd backend/windmill-duckdb-ffi-internal && ./build_dev.sh`. **Two gotchas that a wrong feature set produces:** diff --git a/backend/Cargo.lock b/backend/Cargo.lock index b2139d6d41..bc5b9aa55c 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -17465,7 +17465,6 @@ dependencies = [ "jsonwebtoken 10.4.0", "lazy_static", "libffi-sys", - "libloading", "mappable-rc", "mime_guess", "mysql_async", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 24955a3744..6fc1341d53 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -723,7 +723,6 @@ flume = { version = "0.11.1", features = ["async"] } kube = { version = "1.1.0", features = ["runtime", "derive"] } schemars = "0.8" k8s-openapi = { version = "0.25.0", features = ["latest"] } -libloading = "0.8.8" # Macro-related proc-macro2 = "1.0" diff --git a/backend/windmill-duckdb-ffi-internal/Cargo.lock b/backend/windmill-duckdb-ffi-internal/Cargo.lock index f92bb6fda1..74cd7b8a18 100644 --- a/backend/windmill-duckdb-ffi-internal/Cargo.lock +++ b/backend/windmill-duckdb-ffi-internal/Cargo.lock @@ -460,19 +460,11 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "diffy" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10aec8f7f9393bd6a4f2762be0ceb012d3cbe2478987258cc9960de148561914" -dependencies = [ - "hashbrown 0.17.1", -] - [[package]] name = "duckdb" -version = "1.10505.0" -source = "git+https://github.com/windmill-labs/duckdb-rs?rev=7190adfcf58624e7e9a7a2a1000bac18fed64c19#7190adfcf58624e7e9a7a2a1000bac18fed64c19" +version = "1.10506.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3472b9152d879c5534353b20854e43525b2c78b2dbba156711e9939d30278b4" dependencies = [ "arrow", "cast", @@ -547,12 +539,6 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" -[[package]] -name = "foldhash" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" - [[package]] name = "funty" version = "2.0.0" @@ -644,7 +630,7 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "foldhash 0.1.5", + "foldhash", ] [[package]] @@ -652,9 +638,6 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" -dependencies = [ - "foldhash 0.2.0", -] [[package]] name = "hashlink" @@ -813,11 +796,11 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libduckdb-sys" -version = "1.10505.0" -source = "git+https://github.com/windmill-labs/duckdb-rs?rev=7190adfcf58624e7e9a7a2a1000bac18fed64c19#7190adfcf58624e7e9a7a2a1000bac18fed64c19" +version = "1.10506.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0aac3459f2f7f0221d5be9114fa77ff5877040c8de456d3b986a680579fe202b" dependencies = [ "cc", - "diffy", "flate2", "pkg-config", "serde", diff --git a/backend/windmill-duckdb-ffi-internal/Cargo.toml b/backend/windmill-duckdb-ffi-internal/Cargo.toml index 73e132a2d7..ba78e3522d 100644 --- a/backend/windmill-duckdb-ffi-internal/Cargo.toml +++ b/backend/windmill-duckdb-ffi-internal/Cargo.toml @@ -11,10 +11,9 @@ chrono = "0.4.41" # identical DECIMAL. Building one by hand would mean reimplementing that # f64 → (mantissa, scale) inference. # -# Not crates.io: a fork of duckdb-rs 1.10505.0 carrying two patches to the bundled engine — -# see docs/duckdb-isolation.md for why, and for the exit path back to the crates.io crate. -# Pinned by rev, not branch, so `cargo update` cannot move the engine under us. -duckdb = { git = "https://github.com/windmill-labs/duckdb-rs", rev = "7190adfcf58624e7e9a7a2a1000bac18fed64c19", features = [ +# Pinned exactly, so `cargo update` cannot move the engine (and the prebuilt extensions it +# must match) under us. +duckdb = { version = "=1.10506.0", features = [ "bundled", "rust_decimal", ] } @@ -25,5 +24,6 @@ serde = { version = "1.0", features = ["derive"] } # to avoid re-parsing every row. serde_json = { version = "^1.0.151", features = ["preserve_order", "raw_value"] } -[lib] -crate-type = ["cdylib"] +[[bin]] +name = "windmill_duckdb_runner" +path = "src/bin/windmill_duckdb_runner.rs" diff --git a/backend/windmill-duckdb-ffi-internal/README_DEV.md b/backend/windmill-duckdb-ffi-internal/README_DEV.md index 0c34ebc069..f5c5d61083 100644 --- a/backend/windmill-duckdb-ffi-internal/README_DEV.md +++ b/backend/windmill-duckdb-ffi-internal/README_DEV.md @@ -1,8 +1,8 @@ -This crate is compiled separately because it causes nasty issues when compiled with the deno_core feature flag enabled (lib c++ interactions). +This crate is compiled separately from the backend workspace because the bundled DuckDB's libc++ +clashes with `deno_core`'s when both live in one process. -The main issue was : -Errors in DuckDB always worked correctly, except when attached to a Ducklake and when the deno_core feature flag was set. -For example: +The symptom was opaque errors. DuckDB errors worked, except when attached to a DuckLake with the +`deno_core` feature on: ```sql ATTACH 'ducklake' AS dl; USE dl; @@ -10,6 +10,9 @@ CREATE TABLE IF NOT EXISTS t (x string not null); INSERT INTO t VALUES (NULL); ``` -causes `Constraint Error: NOT NULL constraint failed: t.x` normally, but here we see `Unknown exception in ExecutorTask::Execute`. This opaque errors comes directly from the C++ DuckDB library : https://github.com/duckdb/duckdb/blob/f99fed1e0b16a842573f9dad529f6c170a004f6e/src/parallel/executor_task.cpp#L58 +gave `Unknown exception in ExecutorTask::Execute` instead of `Constraint Error: NOT NULL +constraint failed: t.x`. The opaque error comes from the C++ DuckDB library: +https://github.com/duckdb/duckdb/blob/f99fed1e0b16a842573f9dad529f6c170a004f6e/src/parallel/executor_task.cpp#L58 -To solve this, we compile duckdb separately from the main backend crate and call it with FFI. It has to be loaded dynamically, if it is loaded statically it will still share lib c++ with deno_core and cause issues. +The worker never links this crate: it runs `windmill_duckdb_runner` (`src/bin`) as a separate +process for every job. See `docs/duckdb-execution.md`. diff --git a/backend/windmill-duckdb-ffi-internal/build_dev.sh b/backend/windmill-duckdb-ffi-internal/build_dev.sh index dab70398a8..c334baa6ae 100755 --- a/backend/windmill-duckdb-ffi-internal/build_dev.sh +++ b/backend/windmill-duckdb-ffi-internal/build_dev.sh @@ -31,22 +31,21 @@ else echo "duckdb-ffi: shared cache $CARGO_TARGET_DIR" # The cache key covers only the dependency inputs (Cargo.lock + build.rs), so # the expensive bundled-DuckDB build is shared even when this crate's own - # source differs between checkouts. But the shared dir's uplifted cdylib is a + # source differs between checkouts. But the shared dir's runner binary is a # single fixed-name artifact written by whichever worktree built last. Touch - # our sources so cargo re-links THIS checkout's cdylib and re-uplifts it + # our sources so cargo re-links THIS checkout's runner and re-uplifts it # before we copy — the copied artifact then always matches this worktree, # never a sibling's. The bundled dependency is a separate crate and stays # cached, so this only costs a ~1s relink. find src -type f -exec touch {} + fi -CARGO_NET_GIT_FETCH_WITH_CLI=true cargo build --release -p windmill_duckdb_ffi_internal +cargo build --release -p windmill_duckdb_ffi_internal mkdir -p ../target/debug/ -# Install through a rename rather than writing over the destination: a running worker -# has the cdylib mapped, and rewriting those pages under it kills the process with -# SIGBUS. A rename swaps the directory entry and leaves the old inode alone. -for artifact in "$CARGO_TARGET_DIR/release/"libwindmill_duckdb_ffi_internal.*; do - dest="../target/debug/$(basename "$artifact")" - cp "$artifact" "$dest.tmp" - mv -f "$dest.tmp" "$dest" -done +# The worker looks for the runner next to its own binary. Install through a rename rather +# than writing over the destination: a runner may be executing, and rewriting its pages under +# it kills the process with SIGBUS. A rename swaps the directory entry and leaves the old +# inode alone. +dest="../target/debug/windmill_duckdb_runner" +cp "$CARGO_TARGET_DIR/release/windmill_duckdb_runner" "$dest.tmp" +mv -f "$dest.tmp" "$dest" diff --git a/backend/windmill-duckdb-ffi-internal/src/bin/windmill_duckdb_runner.rs b/backend/windmill-duckdb-ffi-internal/src/bin/windmill_duckdb_runner.rs new file mode 100644 index 0000000000..38052a9537 --- /dev/null +++ b/backend/windmill-duckdb-ffi-internal/src/bin/windmill_duckdb_runner.rs @@ -0,0 +1,122 @@ +//! Runs one DuckDB job, so the engine lives in a process of its own under the worker's job +//! isolation rather than in the worker. +//! +//! The worker writes one JSON [`Request`] to stdin and reads the answer from +//! `Request::result_path`: a first line holding the column order as JSON (empty when there is +//! none), then the collected results, or `ERROR `. +//! +//! stdout/stderr are not protocol: the worker keeps only their tail, and reports it when the +//! runner dies without writing a result. + +use std::io::{Read, Write}; + +use serde::Deserialize; +use windmill_duckdb_ffi_internal::{prepare, run, Arg, Connection}; + +#[derive(Deserialize)] +#[serde(rename_all = "snake_case")] +enum Mode { + Run, + Prepare, + /// `INSTALL` only, into the extension cache the jail later mounts read-only. Never loads + /// anything, so it runs no extension code. + Install, +} + +#[derive(Deserialize)] +struct Request { + mode: Mode, + blocks: Vec, + #[serde(default)] + job_args: Vec, + #[serde(default)] + token: String, + #[serde(default)] + base_internal_url: String, + #[serde(default)] + w_id: String, + #[serde(default)] + memory_limit: String, + #[serde(default)] + temp_directory: String, + #[serde(default)] + max_result_size: usize, + #[serde(default)] + collect_last_only: bool, + #[serde(default)] + collect_first_row_only: bool, + result_path: String, +} + +fn non_empty(s: &str) -> Option { + (!s.is_empty()).then(|| s.to_string()) +} + +fn install(blocks: &[String]) -> Result { + let conn = duckdb::Connection::open_in_memory().map_err(|e| e.to_string())?; + for b in blocks { + conn.execute_batch(b).map_err(|e| e.to_string())?; + } + Ok("[]".to_string()) +} + +/// The result string and the column order as JSON. +fn execute(req: Request) -> (String, Option) { + let conn = Connection { + token: &req.token, + base_internal_url: &req.base_internal_url, + w_id: &req.w_id, + memory_limit: non_empty(&req.memory_limit), + temp_directory: non_empty(&req.temp_directory), + }; + let outcome = match req.mode { + Mode::Install => install(&req.blocks).map(|r| (r, None)), + Mode::Prepare => prepare(&req.blocks, conn).map(|r| (r, None)), + Mode::Run => run( + &req.blocks, + req.job_args, + conn, + req.max_result_size, + req.collect_last_only, + req.collect_first_row_only, + ) + .map(|(r, order)| { + let order = order.map(|o| serde_json::to_string(&o).unwrap_or_else(|_| "[]".into())); + (r, order) + }), + }; + outcome.unwrap_or_else(|e| { + let msg = serde_json::to_string(&e).unwrap_or_else(|_| "\"Unknown DuckDB error\"".into()); + (format!("ERROR {msg}"), None) + }) +} + +fn main() { + let mut input = String::new(); + if let Err(e) = std::io::stdin().read_to_string(&mut input) { + eprintln!("windmill_duckdb_runner: could not read the request: {e}"); + std::process::exit(2); + } + let req: Request = match serde_json::from_str(&input) { + Ok(r) => r, + Err(e) => { + eprintln!("windmill_duckdb_runner: invalid request: {e}"); + std::process::exit(2); + } + }; + drop(input); + + let result_path = req.result_path.clone(); + let (result, column_order) = execute(req); + let write = || -> std::io::Result<()> { + let mut f = std::io::BufWriter::new(std::fs::File::create(&result_path)?); + f.write_all(column_order.as_deref().unwrap_or("").as_bytes())?; + f.write_all(b"\n")?; + f.write_all(result.as_bytes())?; + f.flush() + }; + if let Err(e) = write() { + eprintln!("windmill_duckdb_runner: could not write the result to {result_path}: {e}"); + std::process::exit(2); + } +} diff --git a/backend/windmill-duckdb-ffi-internal/src/lib.rs b/backend/windmill-duckdb-ffi-internal/src/lib.rs index 69820b285f..0de7d3bd79 100644 --- a/backend/windmill-duckdb-ffi-internal/src/lib.rs +++ b/backend/windmill-duckdb-ffi-internal/src/lib.rs @@ -1,9 +1,9 @@ -use std::{ - collections::HashMap, - ffi::{c_char, c_uint, CStr, CString}, - ptr::null_mut, - sync::LazyLock, -}; +//! DuckDB execution for windmill jobs. The worker never loads this crate: it runs +//! `windmill_duckdb_runner` (src/bin), which calls [`run`] / [`prepare`] in a process of its +//! own, under the worker's job isolation. Kept out of the backend workspace because the +//! bundled engine's libc++ clashes with `deno_core`'s in one process. + +use std::{collections::HashMap, sync::LazyLock}; use duckdb::{core::LogicalTypeId, params_from_iter, types::TimeUnit, Row}; use regex::Regex; @@ -11,22 +11,6 @@ use rust_decimal::{prelude::FromPrimitive, Decimal}; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; -// Worker passes "" for "no override" — saves an extra C string nullability dance. -// Returns an owned String so the value outlives the raw pointer's lifetime. -fn ptr_to_opt_str(ptr: *const c_char) -> Result, String> { - if ptr.is_null() { - return Ok(None); - } - let s = unsafe { CStr::from_ptr(ptr) } - .to_str() - .map_err(|e| format!("Invalid string in duckdb ffi: {}", e))?; - Ok(if s.is_empty() { - None - } else { - Some(s.to_owned()) - }) -} - #[derive(Deserialize, Clone, Debug, PartialEq, Default)] pub struct Arg { pub name: String, @@ -34,95 +18,69 @@ pub struct Arg { pub json_value: serde_json::Value, } -// Freeing from the caller side crashes the runtime with jemalloc enabled (EXIT CODE 11 SEGFAULT) -#[unsafe(no_mangle)] -pub extern "C" fn free_cstr(string: *mut c_char) -> () { - if string.is_null() { - return; - } - unsafe { - let _ = CString::from_raw(string); - } +/// Everything a connection is set up with before the script's first block runs. +pub struct Connection<'a> { + /// The job token: the S3/GCS secrets against the workspace's S3 proxy are derived from it. + pub token: &'a str, + pub base_internal_url: &'a str, + pub w_id: &'a str, + pub memory_limit: Option, + /// Where DuckDB spills; the job dir, so the spill is cleaned up with the job. + pub temp_directory: Option, } -#[unsafe(no_mangle)] -pub extern "C" fn get_version() -> c_uint { - // Increment when making breaking changes to the FFI interface. - // The windmill worker will check that the version matches or else refuse to call - // the FFI functions to avoid undefined behavior. - return 3; -} - -#[unsafe(no_mangle)] -pub extern "C" fn run_duckdb_ffi( - query_block_list: *const *const c_char, - query_block_list_count: usize, - job_args: *const c_char, - token: *const c_char, - base_internal_url: *const c_char, - w_id: *const c_char, - memory_limit: *const c_char, - temp_directory: *const c_char, - // 0 means unbounded; the worker sends its own budget-derived cap. +/// Runs the blocks in order on one in-memory connection and returns every collected result, +/// serialized, plus the last statement's column order. `max_result_size` 0 is unbounded. +pub fn run( + blocks: &[String], + job_args: Vec, + conn: Connection<'_>, max_result_size: usize, - column_order_ptr: *mut *mut c_char, collect_last_only: bool, collect_first_row_only: bool, -) -> *mut c_char { - let resource_limits = match (ptr_to_opt_str(memory_limit), ptr_to_opt_str(temp_directory)) { - (Ok(m), Ok(t)) => Ok(ResourceLimits::from_ffi(m, t)), - (Err(e), _) | (_, Err(e)) => Err(e), - }; - let (r, column_order) = match convert_args( - query_block_list, - query_block_list_count, - job_args, - token, - base_internal_url, - w_id, +) -> Result<(String, Option>), String> { + let limits = ResourceLimits::new(conn.memory_limit, conn.temp_directory); + run_duckdb_internal( + blocks.iter().map(String::as_str), + blocks.len(), + job_args_to_values(job_args), + conn.token, + conn.base_internal_url, + conn.w_id, + limits, + max_result_size, + collect_last_only, + collect_first_row_only, ) - .and_then(|args| resource_limits.map(|r| (args, r))) - .and_then( - |((query_block_list, job_args, token, base_internal_url, w_id), limits)| { - run_duckdb_internal( - query_block_list, - query_block_list_count, - job_args, - token, - base_internal_url, - w_id, - limits, - max_result_size, - collect_last_only, - collect_first_row_only, - ) - }, - ) { - Ok(result) => result, - Err(err) => { - let err = serde_json::to_string(&err) - .unwrap_or_else(|_| "Unknown error in duckdb ffi lib".to_string()); - (format!("ERROR {}", err), None) - } - }; +} - unsafe { - if let Some(column_order) = column_order { - let column_order = - serde_json::to_string(&column_order).unwrap_or_else(|_| "[]".to_string()); - let c_column_order = - CString::new(column_order).unwrap_or_else(|_| CString::new("[]").unwrap()); - *column_order_ptr = c_column_order.into_raw(); - } else { - *column_order_ptr = null_mut(); - } - } - // CString::into_raw because it needs to outlive this function call. - // The caller is responsible for freeing the memory through CString::from_raw. - CString::new(r).map(|s| s.into_raw()).unwrap_or_else(|e| { - println!("Failed to allocate error string in duckdb ffi lib: {:?}", e); - null_mut() - }) +/// Describes each non-setup block's result columns without running it. +pub fn prepare(blocks: &[String], conn: Connection<'_>) -> Result { + let limits = ResourceLimits::new(conn.memory_limit, conn.temp_directory); + prepare_duckdb_internal( + blocks.iter().map(String::as_str).collect(), + conn.token, + conn.base_internal_url, + conn.w_id, + limits, + ) +} + +fn job_args_to_values(job_args: Vec) -> HashMap { + job_args + .into_iter() + .map(|arg| { + let duckdb_value = json_value_to_duckdb_value(&arg.json_value, &arg.arg_type) + .unwrap_or_else(|e| { + println!( + "Error converting job_arg {} to duckdb value: {}", + arg.name, e + ); + duckdb::types::Value::Null + }); + (arg.name, duckdb_value) + }) + .collect() } #[derive(Serialize, Debug)] @@ -187,47 +145,6 @@ fn replace_params_with_null(query: &str) -> String { PARAM_RE.replace_all(query, "NULL").to_string() } -#[unsafe(no_mangle)] -pub extern "C" fn prepare_duckdb_ffi( - query_block_list: *const *const c_char, - query_block_list_count: usize, - token: *const c_char, - base_internal_url: *const c_char, - w_id: *const c_char, - memory_limit: *const c_char, - temp_directory: *const c_char, -) -> *mut c_char { - let resource_limits = match (ptr_to_opt_str(memory_limit), ptr_to_opt_str(temp_directory)) { - (Ok(m), Ok(t)) => Ok(ResourceLimits::from_ffi(m, t)), - (Err(e), _) | (_, Err(e)) => Err(e), - }; - let r = match convert_prepare_args( - query_block_list, - query_block_list_count, - token, - base_internal_url, - w_id, - ) - .and_then(|args| resource_limits.map(|r| (args, r))) - .and_then( - |((query_block_list, token, base_internal_url, w_id), limits)| { - prepare_duckdb_internal(query_block_list, token, base_internal_url, w_id, limits) - }, - ) { - Ok(result) => result, - Err(err) => { - let err = serde_json::to_string(&err) - .unwrap_or_else(|_| "Unknown error in duckdb ffi lib".to_string()); - format!("ERROR {}", err) - } - }; - - CString::new(r).map(|s| s.into_raw()).unwrap_or_else(|e| { - println!("Failed to allocate error string in duckdb ffi lib: {:?}", e); - null_mut() - }) -} - #[derive(Clone, Default)] struct ResourceLimits { memory_limit: Option, @@ -236,11 +153,10 @@ struct ResourceLimits { } impl ResourceLimits { - // The spill cap comes from the environment rather than from an FFI parameter: this - // cdylib is versioned separately from the worker and agent workers ship it out of band, - // so a signature change costs an ABI bump and a coordinated redeploy for what is a - // config value. Unset keeps DuckDB's default of 90% of the temp volume's free space. - fn from_ffi(memory_limit: Option, temp_directory: Option) -> Self { + // The spill cap is the worker's DUCKDB_MAX_TEMP_DIRECTORY_SIZE, passed through to the + // runner's environment. Unset keeps DuckDB's default of 90% of the temp volume's free + // space. A script can raise it, as it can write the job dir any other way. + fn new(memory_limit: Option, temp_directory: Option) -> Self { let max_temp_directory_size = std::env::var("DUCKDB_MAX_TEMP_DIRECTORY_SIZE") .ok() .map(|v| v.trim().to_string()) @@ -253,10 +169,9 @@ fn sql_single_quote(s: &str) -> String { s.replace('\'', "''") } -// Bounds memory so DuckDB spills before blowing the cgroup cap and getting the worker -// SIGKILLed, and bounds the spill so one query cannot fill its disk. Spill goes to the job -// dir when set, so it is cleaned up with the job; either way the directory and its cap are -// then locked for the rest of the connection. +// Bounds memory so DuckDB spills before blowing the cgroup cap and getting the job +// SIGKILLed, and sets a default bound on the spill. Spill goes to the job dir when set, so it +// is cleaned up with the job. fn configure_duckdb_resource_limits( conn: &duckdb::Connection, limits: &ResourceLimits, @@ -301,11 +216,7 @@ fn configure_duckdb_resource_limits( ) })?; } - // Freeze both before any script statement can touch them. A remote temp directory makes - // `~TemporaryDirectoryHandle` throw out of a `noexcept` destructor — `std::terminate`, - // taking this in-process worker and every job on it down, uncatchable from Rust — and a - // cap a script can raise bounds nothing. - run("SET lock_temp_directory=true;\n") + Ok(()) } fn setup_duckdb_connection( @@ -353,39 +264,6 @@ fn setup_duckdb_connection( .map_err(|e| format!("Error setting up S3 secret: {}", e.to_string())) } -fn convert_prepare_args<'a>( - query_block_list: *const *const c_char, - query_block_list_count: usize, - token: *const c_char, - base_internal_url: *const c_char, - w_id: *const c_char, -) -> Result<(Vec<&'a str>, &'a str, &'a str, &'a str), String> { - let query_block_list = unsafe { - std::slice::from_raw_parts(query_block_list, query_block_list_count) - .iter() - .map(|q| { - CStr::from_ptr(*q).to_str().unwrap_or_else(|e| { - println!( - "Invalid query_block string pointer in duckdb ffi: {}", - e.to_string() - ); - "Invalid query_block string pointer in duckdb ffi" - }) - }) - .collect::>() - }; - let token = unsafe { CStr::from_ptr(token) } - .to_str() - .map_err(|e| format!("Invalid token string: {}", e.to_string()))?; - let base_internal_url = unsafe { CStr::from_ptr(base_internal_url) } - .to_str() - .map_err(|e| format!("Invalid base_internal_url string: {}", e.to_string()))?; - let w_id = unsafe { CStr::from_ptr(w_id) } - .to_str() - .map_err(|e| format!("Invalid w_id string: {}", e.to_string()))?; - Ok((query_block_list, token, base_internal_url, w_id)) -} - fn prepare_duckdb_internal( query_block_list: Vec<&str>, token: &str, @@ -453,67 +331,6 @@ fn prepare_duckdb_internal( serde_json::to_string(&results).map_err(|e| e.to_string()) } -fn convert_args<'a>( - query_block_list: *const *const c_char, - query_block_list_count: usize, - job_args: *const c_char, - token: *const c_char, - base_internal_url: *const c_char, - w_id: *const c_char, -) -> Result< - ( - impl Iterator, - HashMap, - &'a str, - &'a str, - &'a str, - ), - String, -> { - let query_block_list = unsafe { - std::slice::from_raw_parts(query_block_list, query_block_list_count) - .iter() - .map(|q| { - CStr::from_ptr(*q).to_str().unwrap_or_else(|e| { - println!( - "Invalid query_block string pointer in duckdb ffi: {}", - e.to_string() - ); - "Invalid query_block string pointer in duckdb ffi" - }) - }) - }; - let job_args_str = unsafe { CStr::from_ptr(job_args) } - .to_str() - .map_err(|e| format!("Invalid job_args string: {}", e.to_string()))?; - let job_args: Vec = serde_json::from_str(job_args_str) - .map_err(|e| format!("Invalid job_args JSON: {}", e.to_string()))?; - let job_args: HashMap = job_args - .into_iter() - .map(|arg| { - let duckdb_value = json_value_to_duckdb_value(&arg.json_value, &arg.arg_type) - .unwrap_or_else(|e| { - println!( - "Error converting job_arg {} to duckdb value: {}", - arg.name, e - ); - duckdb::types::Value::Null - }); - (arg.name, duckdb_value) - }) - .collect(); - let token = unsafe { CStr::from_ptr(token) } - .to_str() - .map_err(|e| format!("Invalid token string: {}", e.to_string()))?; - let base_internal_url = unsafe { CStr::from_ptr(base_internal_url) } - .to_str() - .map_err(|e| format!("Invalid base_internal_url string: {}", e.to_string()))?; - let w_id = unsafe { CStr::from_ptr(w_id) } - .to_str() - .map_err(|e| format!("Invalid w_id string: {}", e.to_string()))?; - Ok((query_block_list, job_args, token, base_internal_url, w_id)) -} - // TODO : Better error return to leverage different error kinds on the worker side fn run_duckdb_internal<'a>( query_block_list: impl Iterator, @@ -1230,8 +1047,8 @@ mod temporal_json_tests { // Numbers too wide for a JSON number are rendered as strings. DECIMAL runs to // 38 digits and UHUGEINT to 2^128-1; rendering either through a type that - // cannot hold it aborts the whole worker rather than erroring, because the - // panic escapes an `extern "C"` frame and those cannot unwind. + // cannot hold it panics, which kills the runner and fails the job with no + // engine message to show for it. #[test] fn wide_numbers_render_as_strings_without_losing_precision() { let conn = duckdb::Connection::open_in_memory().unwrap(); @@ -1677,214 +1494,3 @@ fn string_to_duckdb_time(s: &str) -> Result { time.num_seconds_from_midnight() as i64, )) } - -// The bundled engine is not stock (docs/duckdb-isolation.md). These guard the ways an engine -// bump can drop the patches without anything else noticing. -#[cfg(test)] -mod patched_engine_tests { - use super::*; - - /// Too large to be answered within the `memory_limit` `spill_attempt` sets, so the engine - /// has to reach the temp directory to finish it. - const SPILLING_QUERY: &str = "CREATE TABLE t AS SELECT * FROM range(1000000)"; - - /// Every connection is locked at setup, so no script statement can move the temp - /// directory onto a remote filesystem — where the spill fails and the - /// `~TemporaryDirectoryHandle` cleaning up throws out of a `noexcept` destructor, - /// aborting the worker and every job colocated on it. - #[test] - fn configured_connections_cannot_have_their_temp_directory_moved() { - let temp_dir = std::env::temp_dir().join(format!( - "windmill_duckdb_locked_{}_{:?}", - std::process::id(), - std::thread::current().id() - )); - std::fs::create_dir_all(&temp_dir).unwrap(); - - let conn = duckdb::Connection::open_in_memory().unwrap(); - configure_duckdb_resource_limits( - &conn, - &ResourceLimits { - memory_limit: Some("2MB".to_string()), - temp_directory: Some(temp_dir.display().to_string()), - max_temp_directory_size: None, - }, - ) - .unwrap(); - - for stmt in [ - "SET temp_directory='s3://bucket/tmp'", - "SET temp_directory TO 's3://bucket/tmp'", - "PRAGMA temp_directory='s3://bucket/tmp'", - "RESET temp_directory", - ] { - let err = conn - .execute_batch(stmt) - .expect_err(&format!("engine accepted `{stmt}` on a locked connection")); - assert!( - err.to_string().contains("temp directory has been locked"), - "unexpected refusal for `{stmt}`: {err}" - ); - } - - // The lock must not cost the spilling it exists to protect. - let spilled = conn.execute_batch(SPILLING_QUERY); - std::fs::remove_dir_all(&temp_dir).ok(); - spilled.expect("a locked temp directory must still be usable for spilling"); - } - - /// `DUCKDB_MAX_TEMP_DIRECTORY_SIZE` is the only thing standing between one query and the - /// worker's whole disk, so pin that a configured connection really does stop at it, names - /// the setting when it does, and leaves nothing behind for the next job to trip over. - #[test] - fn spilling_past_the_configured_cap_fails_the_query_and_cleans_up() { - let temp_dir = std::env::temp_dir().join(format!( - "windmill_duckdb_capped_{}_{:?}", - std::process::id(), - std::thread::current().id() - )); - std::fs::create_dir_all(&temp_dir).unwrap(); - - let spill_under_cap = |cap: &str| { - let conn = duckdb::Connection::open_in_memory().unwrap(); - configure_duckdb_resource_limits( - &conn, - &ResourceLimits { - memory_limit: Some("2MB".to_string()), - temp_directory: Some(temp_dir.display().to_string()), - max_temp_directory_size: Some(cap.to_string()), - }, - ) - .unwrap(); - conn.execute_batch(SPILLING_QUERY) - .map_err(|e| e.to_string()) - }; - - let err = spill_under_cap("1MB").expect_err("the cap did not stop the spill"); - // Verbatim because the worker matches on it (`SPILL_CAP_ENGINE_MARKER` in - // `duckdb_executor.rs`) to correct the engine's advice to raise the cap, which the - // lock refuses. An engine bump that rewords this has to update that too. - assert!( - err.contains("This limit was set by the 'max_temp_directory_size' setting"), - "the refusal must name the setting an operator can act on, got: {err}" - ); - let leftover: Vec<_> = std::fs::read_dir(&temp_dir) - .unwrap() - .filter_map(|e| e.ok().map(|e| e.path())) - .collect(); - assert!( - leftover.is_empty(), - "a query killed by the cap must not leave its spill behind: {leftover:?}" - ); - - // DuckDB compresses temp files, so the failure above is only evidence of the cap if - // the same query gets through once the cap is out of the way. - let generous = spill_under_cap("1GB"); - std::fs::remove_dir_all(&temp_dir).ok(); - generous.expect("the query must still spill and finish under a cap that fits it"); - } - - fn spill_attempt(temp_dir: &std::path::Path, lock: bool) -> Result<(), String> { - let conn = duckdb::Connection::open_in_memory().unwrap(); - conn.execute_batch(&format!( - "SET threads=2; SET memory_limit='2MB'; SET temp_directory='{}';", - temp_dir.display() - )) - .unwrap(); - if lock { - conn.execute_batch("SET lock_temp_directory=true;") - .map_err(|e| format!("lock_temp_directory rejected: {e}"))?; - } - conn.execute_batch("SET disabled_filesystems='LocalFileSystem';") - .unwrap(); - conn.execute_batch(SPILLING_QUERY) - .map_err(|e| e.to_string()) - } - - #[test] - fn locking_the_temp_directory_keeps_spilling_available() { - let temp_dir = std::env::temp_dir().join(format!( - "windmill_duckdb_spill_{}_{:?}", - std::process::id(), - std::thread::current().id() - )); - std::fs::create_dir_all(&temp_dir).unwrap(); - - // Without the lock the fence reaches the buffer manager too, which is both the - // stock engine's behaviour and the proof that this query really does spill — - // otherwise the assertion below would pass on an unpatched engine. - let unlocked = spill_attempt(&temp_dir, false).expect_err("query did not need to spill"); - assert!( - unlocked.contains("LocalFileSystem has been disabled"), - "expected the fence to stop the spill, got: {unlocked}" - ); - - let locked = spill_attempt(&temp_dir, true); - std::fs::remove_dir_all(&temp_dir).ok(); - locked.expect("a locked temp directory must stay usable behind disabled_filesystems"); - } - - /// The exemption above is only sound while neither the spill's destination nor its size - /// can be changed: repointing the directory is a write primitive through the very fence - /// it is exempt from, and raising the cap leaves the disk unbounded. A rebase can drop - /// either half while leaving the spill test green, so pin both. - #[test] - fn locking_the_temp_directory_makes_it_immutable() { - let conn = duckdb::Connection::open_in_memory().unwrap(); - conn.execute_batch("SET temp_directory='/tmp/windmill_duckdb_lock_probe';") - .unwrap(); - conn.execute_batch("SET lock_temp_directory=true;") - .expect("the patched engine must accept lock_temp_directory"); - - for stmt in [ - "SET temp_directory='/tmp/windmill_duckdb_elsewhere';", - "RESET temp_directory;", - "SET max_temp_directory_size='100GB';", - "RESET max_temp_directory_size;", - "SET lock_temp_directory=false;", - "RESET lock_temp_directory;", - ] { - let err = conn.execute_batch(stmt).expect_err(&format!( - "`{stmt}` must be refused once the directory is locked" - )); - assert!( - err.to_string().contains("locked"), - "expected a lock refusal for `{stmt}`, got: {err}" - ); - } - } - - // The engine loads the *prebuilt* extensions from extensions.duckdb.org, which are - // compiled against the struct layouts of the release it claims to be. A patch that - // moves a member those extensions reach does not fail — the extension reads the - // neighbouring field and LOAD blocks forever — so this is a timeout, not an assertion. - #[test] - fn prebuilt_extensions_still_load() { - let conn = duckdb::Connection::open_in_memory().unwrap(); - // Fetching the extension needs egress, which loading it back — the part this guards — - // does not. Skipping keeps a developer offline from seeing a failure they can't act on, - // but never in CI, where a skip and a pass would be indistinguishable. - if let Err(e) = conn.execute_batch("INSTALL httpfs;") { - assert!( - std::env::var_os("CI").is_none(), - "could not install httpfs, so the extension-load guard did not run: {e}" - ); - eprintln!("skipping: INSTALL httpfs failed ({e})"); - return; - } - let (tx, rx) = std::sync::mpsc::channel(); - std::thread::spawn(move || { - tx.send( - conn.execute_batch("LOAD httpfs;") - .map_err(|e| e.to_string()), - ) - }); - match rx.recv_timeout(std::time::Duration::from_secs(120)) { - Ok(r) => r.expect("could not load httpfs"), - Err(_) => panic!( - "LOAD httpfs did not return: the patched engine's struct layout no longer matches \ - the release the prebuilt extensions were built against" - ), - } - } -} diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 57b292f8f2..68728a9713 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -39,7 +39,7 @@ nu = ["dep:windmill-parser-nu"] java = ["dep:windmill-parser-java"] ruby = ["dep:windmill-parser-ruby"] rlang = ["dep:windmill-parser-r"] -duckdb = ["dep:libloading"] +duckdb = [] quickjs = ["windmill-jseval/quickjs", "windmill-queue/quickjs"] bedrock = ["windmill-ai/bedrock"] @@ -143,7 +143,6 @@ backon.workspace = true pep440_rs.workspace = true process-wrap.workspace = true async-once-cell.workspace = true -libloading = { workspace = true, optional = true } opentelemetry-proto = { workspace = true, optional = true } opentelemetry = { workspace = true, optional = true } diff --git a/backend/windmill-worker/nsjail/run.duckdb.config.proto b/backend/windmill-worker/nsjail/run.duckdb.config.proto new file mode 100644 index 0000000000..426f3d583d --- /dev/null +++ b/backend/windmill-worker/nsjail/run.duckdb.config.proto @@ -0,0 +1,145 @@ +name: "duckdb run" + +mode: ONCE +hostname: "duckdb" +log_level: ERROR +# The executor's own job timeout kills the jail; nsjail adds none of its own. +time_limit: 0 + +# DuckDB bounds its memory through memory_limit, and the worker bounds its disk use (see the +# work dir mount). rlimit_as would break it: jemalloc reserves address space well past what it +# ever touches. +disable_rl: true + +cwd: "/tmp" + +clone_newnet: false +clone_newuser: {CLONE_NEWUSER} + +skip_setsid: true +keep_caps: false +keep_env: true +mount_proc: true + +mount { + src: "/lib" + dst: "/lib" + is_bind: true +} + +mount { + src: "/lib64" + dst: "/lib64" + is_bind: true + mandatory: false +} + +mount { + src: "/usr" + dst: "/usr" + is_bind: true +} + +mount { + src: "/proc/self/fd" + dst: "/dev/fd" + is_symlink: true + mandatory: false +} + +mount { + src: "/dev/null" + dst: "/dev/null" + is_bind: true + rw: true +} + +mount { + src: "/dev/random" + dst: "/dev/random" + is_bind: true +} + +mount { + src: "/dev/urandom" + dst: "/dev/urandom" + is_bind: true +} + +# DuckDB sizes its thread pool from the cgroup's CPU quota. +mount { + src: "/sys/fs/cgroup" + dst: "/sys/fs/cgroup" + is_bind: true + mandatory: false +} + +mount { + src: "/etc" + dst: "/etc" + is_bind: true +} + +# Container runtimes bind exactly these 3 files as separate submounts over +# /etc; nsjail's ro remount of /etc is non-recursive so they stay writable. +# Load-bearing -- do not remove as redundant with the /etc bind above. +mount { + src: "/etc/resolv.conf" + dst: "/etc/resolv.conf" + is_bind: true + mandatory: false +} + +mount { + src: "/etc/hosts" + dst: "/etc/hosts" + is_bind: true + mandatory: false +} + +mount { + src: "/etc/hostname" + dst: "/etc/hostname" + is_bind: true + mandatory: false +} + +{TMP_MOUNT_BLOCK} + +# At its host path, so every path the worker wrote into the blocks (Postgres root +# certificates) resolves unchanged. Read-only: the one writable place on the host is the +# runner's work dir below, which the worker caps at DUCKDB_MAX_TEMP_DIRECTORY_SIZE. +mount { + src: "{JOB_DIR}" + dst: "{JOB_DIR}" + is_bind: true +} + +# The spill directory and the result file. +mount { + src: "{JOB_DIR}/duckdb" + dst: "{JOB_DIR}/duckdb" + is_bind: true + rw: true +} + +mount { + src: "{RUNNER_PATH}" + dst: "/tmp/windmill_duckdb_runner" + is_bind: true +} + +# Read-only: the worker installs into it before the jail starts, so a job can load what it +# named but can neither add to nor replace what the next job loads. +mount { + src: "{EXTENSION_DIR}" + dst: "/tmp/.duckdb/extensions" + is_bind: true + mandatory: false +} + +{BIGQUERY_CREDENTIALS_MOUNT} + +envar: "HOME=/tmp" + +#{DEV} diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 513d30d181..05c35fce8d 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -1,10 +1,6 @@ -use std::cell::RefCell; use std::env; -use std::ffi::{c_char, c_uint, CStr, CString}; -use std::ptr::NonNull; use std::sync::{Arc, Mutex}; -use libloading::{Library, Symbol}; use serde::Serialize; use serde_json::value::RawValue; use serde_json::{json, Value}; @@ -14,7 +10,8 @@ use windmill_common::utils::sanitize_string_from_password; use windmill_common::worker::{get_memory, to_raw_value, Connection, SqlResultCollectionStrategy}; use windmill_common::workspaces::{ get_datatable_resource_from_db, get_ducklake_from_db_unchecked, - strip_fork_reserved_attach_args, DatatableAccess, DatatableConnectAs, DucklakeCatalogResourceType, + strip_fork_reserved_attach_args, DatatableAccess, DatatableConnectAs, + DucklakeCatalogResourceType, }; use windmill_common::PgDatabase; use windmill_object_store::S3_PROXY_LAST_ERRORS_CACHE; @@ -24,6 +21,7 @@ use windmill_types::s3::S3Object; use crate::agent_workers::{get_datatable_resource_from_agent_http, get_ducklake_from_agent_http}; use crate::common::{build_args_values, get_reserved_variables, OccupancyMetrics}; +use crate::duckdb_runner::{RequestMode, RunnerMode, RunnerRequest}; use crate::handle_child::run_future_with_polling_update_job_poller; #[cfg(feature = "mysql")] use crate::mysql_executor::MysqlDatabase; @@ -405,23 +403,19 @@ async fn compute_sync_alter_ddl( base_internal_url: String, w_id: String, job_dir: String, + runner: RunnerMode, ) -> Result { - let n = probe_blocks.len(); - let (result, _) = tokio::task::spawn_blocking(move || { - run_duckdb_ffi_safe( - probe_blocks.iter().map(String::as_str), - n, - job_args, - &token, - &base_internal_url, - &w_id, - &job_dir, - SqlResultCollectionStrategy::LastStatementAllRows, - ) - }) + let (result, _) = run_duckdb( + runner, + probe_blocks, + job_args, + token, + base_internal_url, + w_id, + job_dir, + SqlResultCollectionStrategy::LastStatementAllRows, + ) .await - .map_err(|e| Error::from(to_anyhow(e))) - .and_then(|r| r) .map_err(|e| { Error::ExecutionErr(format!( "on_schema_change=sync: the schema pre-pass probe failed ({e}); refusing to write" @@ -1328,6 +1322,7 @@ pub async fn do_duckdb( let token = client.token.clone(); let hidden_passwords = Arc::new(Mutex::new(Vec::::new())); + let runner = crate::duckdb_runner::runner_mode(); let result_f = async { let mut hidden_passwords = hidden_passwords.clone(); @@ -1522,10 +1517,6 @@ pub async fn do_duckdb( probe_blocks, ) .await?; - // The probe opens its own DuckDB connection, so it needs the same - // connection policy as the main query below. - #[cfg(feature = "private")] - let probe_blocks = crate::duckdb_isolation_ee::apply_duckdb_isolation(probe_blocks)?; let alter_ddl = compute_sync_alter_ddl( probe_blocks, job_args.clone(), @@ -1534,6 +1525,7 @@ pub async fn do_duckdb( client.base_internal_url.clone(), job.workspace_id.clone(), job_dir.to_string(), + runner, ) .await?; query = query.replace( @@ -1610,46 +1602,34 @@ pub async fn do_duckdb( ) .await?; - #[cfg(feature = "private")] - let query_block_list = - crate::duckdb_isolation_ee::apply_duckdb_isolation(query_block_list)?; let base_internal_url = client.base_internal_url.clone(); let w_id = job.workspace_id.clone(); let job_dir = job_dir.to_string(); if annotations.prepare { - let result = tokio::task::spawn_blocking(move || { - prepare_duckdb_ffi_safe( - query_block_list.iter().map(String::as_str), - &token, - &base_internal_url, - &w_id, - &job_dir, - ) - }) - .await - .map_err(|e| Error::from(to_anyhow(e))) - .and_then(|r| r)?; - - return Ok(result); + return prepare_duckdb( + runner, + query_block_list, + token.clone(), + base_internal_url, + w_id, + job_dir, + ) + .await; } - let result = tokio::task::spawn_blocking(move || { - run_duckdb_ffi_safe( - query_block_list.iter().map(String::as_str), - query_block_list.len(), - job_args, - &token, - &base_internal_url, - &w_id, - &job_dir, - collection_strategy, - ) - }) - .await - .map_err(|e| Error::from(to_anyhow(e))) - .and_then(|r| r); + let result = run_duckdb( + runner, + query_block_list, + job_args, + token.clone(), + base_internal_url, + w_id, + job_dir, + collection_strategy, + ) + .await; let (result, column_order) = match result { Ok(r) => r, @@ -1857,8 +1837,6 @@ pub async fn do_duckdb( match result { Ok(result) => Ok(result), Err(e) => { - #[cfg(feature = "private")] - let e = crate::duckdb_isolation_ee::explain_error(e); // Passwords might appear in the error message — and, for the // structured data-test failure, in sampled row data read from an // attached database — so every outgoing error is sanitized here. @@ -1903,111 +1881,6 @@ pub async fn do_duckdb( } } -thread_local! { - static DUCKDB_FFI_LIB_SINGLETON: RefCell<*const DuckDbFfiLib> = RefCell::new(std::ptr::null()); -} - -struct DuckDbFfiLib { - run_duckdb_ffi: Symbol< - 'static, - unsafe extern "C" fn( - query_block_list: *const *const c_char, - query_block_list_count: usize, - job_args: *const c_char, - token: *const c_char, - base_internal_url: *const c_char, - w_id: *const c_char, - memory_limit: *const c_char, - temp_directory: *const c_char, - max_result_size: usize, - column_order_ptr: *mut *mut c_char, - collect_last_only: bool, - collect_first_row_only: bool, - ) -> *mut c_char, - >, - prepare_duckdb_ffi: Option< - Symbol< - 'static, - unsafe extern "C" fn( - query_block_list: *const *const c_char, - query_block_list_count: usize, - token: *const c_char, - base_internal_url: *const c_char, - w_id: *const c_char, - memory_limit: *const c_char, - temp_directory: *const c_char, - ) -> *mut c_char, - >, - >, - free_cstr: Symbol<'static, unsafe extern "C" fn(string: *mut c_char) -> ()>, -} - -impl DuckDbFfiLib { - fn get_singleton() -> Result<&'static DuckDbFfiLib> { - DUCKDB_FFI_LIB_SINGLETON.with(|cell| unsafe { - let mut singleton = cell.borrow_mut(); - if singleton.is_null() { - let lib = DuckDbFfiLib::init()?; - let boxed_lib = Box::new(lib); - let lib_ptr = Box::leak(boxed_lib); - *singleton = lib_ptr as *const _; - Ok(NonNull::new_unchecked(*singleton as *mut DuckDbFfiLib).as_ref()) - } else { - Ok(&**singleton) - } - }) - } - fn init() -> Result { - let lib = unsafe { - Library::new(if cfg!(target_os = "macos") { - "libwindmill_duckdb_ffi_internal.dylib" - } else if cfg!(target_os = "windows") { - "windmill_duckdb_ffi_internal.dll" - } else { - "libwindmill_duckdb_ffi_internal.so" - }) - .map_err(|e| { - Error::InternalErr(format!( - "Could not init duckdb. Make sure you have the latest windmill_duckdb_ffi_lib.{} alongside your binary : https://github.com/windmill-labs/windmill/releases \n{}", - if cfg!(target_os = "macos") { "dylib" } - else if cfg!(target_os = "windows") { "dll" } - else { "so" }, - e.to_string() - )) - })? - }; - - let lib = Box::leak(Box::new(lib)); - - // Version mismatch should only be possible on Windows agent workers - // We check for it because FFI interface mismatch will cause undefined behavior / crashes - unsafe { - let expected_version: c_uint = 3; - let get_version: Symbol<'static, unsafe extern "C" fn() -> c_uint> = - lib.get(b"get_version") - .map_err(|e| return Error::ExecutionErr(format!("Could not find get_version in the duckdb ffi library. If you are not using docker, consider manually upgrading windmill_duckdb_ffi_lib. {}", e.to_string())))?; - let actual_version = get_version(); - if actual_version < expected_version { - return Err(Error::InternalErr( - format!("Incompatible duckdb ffi library version. Expected: {expected_version}, actual: {actual_version}. Please update to the latest windmill_duckdb_ffi_lib."), - )); - } else if actual_version > expected_version { - return Err(Error::InternalErr( - format!("Incompatible duckdb ffi library version. Expected: {expected_version}, actual: {actual_version}. Please upgrade your worker to the latest windmill version."), - )); - } - } - - let prepare_duckdb_ffi = unsafe { lib.get(b"prepare_duckdb_ffi").ok() }; - - Ok(DuckDbFfiLib { - run_duckdb_ffi: unsafe { lib.get(b"run_duckdb_ffi").map_err(to_anyhow)? }, - prepare_duckdb_ffi, - free_cstr: unsafe { lib.get(b"free_cstr").map_err(to_anyhow)? }, - }) - } -} - // 20% headroom for Rust runtime + DuckDB's untracked allocations. Mirrors // DuckDB's own default ratio, but applied to the worker's cgroup budget // instead of host RAM. @@ -2037,123 +1910,29 @@ fn cgroup_bytes_to_duckdb_memory_limit(bytes: i64) -> Option { Some(format!("{}MiB", mib.max(64))) } -// Read backend/windmill-duckdb-ffi-internal/README_DEV.md for details about why we use FFI -// The FFI returns errors as `ERROR ` (the message is -// serde_json::to_string'd). Decode the JSON string back so multi-line errors -// render with real newlines, not escaped `\n` inside wrapping quotes; fall back -// to the raw slice if it is somehow not a JSON string. +// The runner reports an engine error as `ERROR `. Decode the JSON string +// back so multi-line errors render with real newlines, not escaped `\n` inside wrapping quotes; +// fall back to the raw slice if it is somehow not a JSON string. fn decode_ffi_error(result_str: &str) -> String { let raw = result_str.strip_prefix("ERROR ").unwrap_or(result_str); - let msg = serde_json::from_str::(raw).unwrap_or_else(|_| raw.to_string()); - // Mirrors the filter the FFI applies to the same variable (`ResourceLimits::from_ffi`); - // they live in separate crates, so the rule is stated twice and must stay one rule. - let configured = env::var("DUCKDB_MAX_TEMP_DIRECTORY_SIZE").is_ok_and(|v| !v.trim().is_empty()); - match spill_cap_hint(&msg, configured) { - Some(hint) => format!("{msg}\n{hint}"), - None => msg, - } + serde_json::from_str::(raw).unwrap_or_else(|_| raw.to_string()) } -/// DuckDB's own wording when a query outgrows the spill cap. Asserted against real engine -/// output by `spilling_past_the_configured_cap_fails_the_query_and_cleans_up` in -/// `windmill-duckdb-ffi-internal`, so an engine bump that rewords it fails there rather than -/// silently dropping the hint. -const SPILL_CAP_ENGINE_MARKER: &str = "This limit was set by the 'max_temp_directory_size' setting"; - -/// Split out of [`decode_ffi_error`] so the branch is testable without mutating the process -/// environment. The engine emits the same text for a configured cap and for its own default, -/// so which limit was hit has to come from the worker's config, not from the message. -fn spill_cap_hint(msg: &str, configured: bool) -> Option<&'static str> { - if !msg.contains(SPILL_CAP_ENGINE_MARKER) { - return None; - } - Some(if configured { - SPILL_CAP_CONFIGURED - } else { - SPILL_CAP_DEFAULT - }) -} - -/// Both hints open by contradicting the engine, whose message ends by suggesting -/// `PRAGMA max_temp_directory_size='10GiB'`: every windmill connection refuses that, since the -/// cap is frozen along with the temp directory, so following it yields a second, -/// unrelated-looking permission error. True either way, because the lock is unconditional. -const SPILL_CAP_CONFIGURED: &str = "\nThe PRAGMA suggested above is refused: this connection's \ - temp directory and the cap on it are frozen, so a script cannot raise the limit. That limit \ - is the worker's DUCKDB_MAX_TEMP_DIRECTORY_SIZE. Spill less (fewer threads, a narrower scan, \ - aggregating before the join), or ask an operator to raise it."; - -/// Nothing configured a cap, so this is DuckDB's own default and the worker is genuinely low on -/// disk. Naming the env var here would send the reader after a knob nobody set. -const SPILL_CAP_DEFAULT: &str = "\nThe PRAGMA suggested above is refused: this connection's temp \ - directory and the cap on it are frozen, so a script cannot raise the limit. No cap is \ - configured on this worker, so that limit is DuckDB's default of 90% of the free space on \ - its temp volume: the worker is low on disk. Free space on it, or spill less (fewer threads, \ - a narrower scan, aggregating before the join)."; - -fn run_duckdb_ffi_safe<'a>( - query_block_list: impl Iterator, +/// Turns what the runner returned into the job result. +fn finish_run( + result_str: String, + column_order: Option, query_block_list_count: usize, - job_args: Vec, - token: &str, - base_internal_url: &str, - w_id: &str, - job_dir: &str, collection_strategy: SqlResultCollectionStrategy, ) -> Result<(Box, Option>)> { - let query_block_list = query_block_list - .map(|s| { - CString::new(s).map_err(|e| { - Error::ExecutionErr(format!("Failed CString conversion: {}", e.to_string())) - }) - }) - .collect::>>()?; - let query_block_list = query_block_list - .iter() - .map(|s| s.as_ptr()) - .collect::>(); - let job_args = serde_json::to_string(&job_args).map_err(to_anyhow)?; - - let job_args = CString::new(job_args).map_err(to_anyhow)?; - let token = CString::new(token).map_err(to_anyhow)?; - let base_internal_url = CString::new(base_internal_url).map_err(to_anyhow)?; - let w_id = CString::new(w_id).map_err(to_anyhow)?; - let memory_limit = - CString::new(resolve_duckdb_memory_limit().unwrap_or_default()).map_err(to_anyhow)?; - let temp_directory = CString::new(job_dir).map_err(to_anyhow)?; - - let run_duckdb_ffi = &DuckDbFfiLib::get_singleton()?.run_duckdb_ffi; - let free_cstr = &DuckDbFfiLib::get_singleton()?.free_cstr; - let mut column_order: *mut c_char = std::ptr::null_mut(); - let result_str = unsafe { - let ptr = run_duckdb_ffi( - query_block_list.as_ptr(), - query_block_list_count, - job_args.as_ptr(), - token.as_ptr(), - base_internal_url.as_ptr(), - w_id.as_ptr(), - memory_limit.as_ptr(), - temp_directory.as_ptr(), - *MAX_SQL_RESULT_SIZE, - &mut column_order, - collection_strategy.collect_last_statement_only(query_block_list_count), - collection_strategy.collect_first_row_only(), - ); - let str = CStr::from_ptr(ptr).to_string_lossy().to_string(); - free_cstr(ptr); - str - }; - - let column_order = if column_order.is_null() - || !collection_strategy.collect_last_statement_only(query_block_list_count) - || collection_strategy.collect_scalar() - { - None - } else { - let str = unsafe { CStr::from_ptr(column_order).to_string_lossy().to_string() }; - unsafe { free_cstr(column_order) }; - Some(serde_json::from_str::>(&str)?) + let column_order = match column_order { + Some(str) + if collection_strategy.collect_last_statement_only(query_block_list_count) + && !collection_strategy.collect_scalar() => + { + Some(serde_json::from_str::>(&str)?) + } + _ => None, }; if result_str.starts_with("ERROR") { @@ -2171,55 +1950,7 @@ fn run_duckdb_ffi_safe<'a>( } } -fn prepare_duckdb_ffi_safe<'a>( - query_block_list: impl Iterator, - token: &str, - base_internal_url: &str, - w_id: &str, - job_dir: &str, -) -> Result> { - let query_block_list = query_block_list - .map(|s| { - CString::new(s).map_err(|e| { - Error::ExecutionErr(format!("Failed CString conversion: {}", e.to_string())) - }) - }) - .collect::>>()?; - let query_block_list = query_block_list - .iter() - .map(|s| s.as_ptr()) - .collect::>(); - - let token = CString::new(token).map_err(to_anyhow)?; - let base_internal_url = CString::new(base_internal_url).map_err(to_anyhow)?; - let w_id = CString::new(w_id).map_err(to_anyhow)?; - let memory_limit = - CString::new(resolve_duckdb_memory_limit().unwrap_or_default()).map_err(to_anyhow)?; - let temp_directory = CString::new(job_dir).map_err(to_anyhow)?; - - let lib = DuckDbFfiLib::get_singleton()?; - let prepare_fn = lib.prepare_duckdb_ffi.as_ref().ok_or_else(|| { - Error::InternalErr( - "prepare_duckdb_ffi not available in duckdb ffi library. Please update to the latest windmill_duckdb_ffi_lib.".to_string(), - ) - })?; - let free_cstr = &lib.free_cstr; - - let result_str = unsafe { - let ptr = prepare_fn( - query_block_list.as_ptr(), - query_block_list.len(), - token.as_ptr(), - base_internal_url.as_ptr(), - w_id.as_ptr(), - memory_limit.as_ptr(), - temp_directory.as_ptr(), - ); - let str = CStr::from_ptr(ptr).to_string_lossy().to_string(); - free_cstr(ptr); - str - }; - +fn finish_prepare(result_str: String) -> Result> { if result_str.starts_with("ERROR") { Err(Error::ExecutionErr(decode_ffi_error(&result_str))) } else { @@ -2227,6 +1958,70 @@ fn prepare_duckdb_ffi_safe<'a>( } } +async fn run_duckdb( + runner: RunnerMode, + query_block_list: Vec, + job_args: Vec, + token: String, + base_internal_url: String, + w_id: String, + job_dir: String, + collection_strategy: SqlResultCollectionStrategy, +) -> Result<(Box, Option>)> { + let n = query_block_list.len(); + let job_args = serde_json::value::to_raw_value(&job_args).map_err(to_anyhow)?; + let memory_limit = resolve_duckdb_memory_limit().unwrap_or_default(); + let out = crate::duckdb_runner::run( + runner, + &RunnerRequest { + mode: RequestMode::Run, + blocks: &query_block_list, + job_args: Some(&job_args), + token: &token, + base_internal_url: &base_internal_url, + w_id: &w_id, + memory_limit: &memory_limit, + temp_directory: &job_dir, + max_result_size: *MAX_SQL_RESULT_SIZE, + collect_last_only: collection_strategy.collect_last_statement_only(n), + collect_first_row_only: collection_strategy.collect_first_row_only(), + }, + &job_dir, + ) + .await?; + finish_run(out.result, out.column_order, n, collection_strategy) +} + +async fn prepare_duckdb( + runner: RunnerMode, + query_block_list: Vec, + token: String, + base_internal_url: String, + w_id: String, + job_dir: String, +) -> Result> { + let memory_limit = resolve_duckdb_memory_limit().unwrap_or_default(); + let out = crate::duckdb_runner::run( + runner, + &RunnerRequest { + mode: RequestMode::Prepare, + blocks: &query_block_list, + job_args: None, + token: &token, + base_internal_url: &base_internal_url, + w_id: &w_id, + memory_limit: &memory_limit, + temp_directory: &job_dir, + max_result_size: 0, + collect_last_only: false, + collect_first_row_only: false, + }, + &job_dir, + ) + .await?; + finish_prepare(out.result) +} + struct ParsedAttachDbResource<'a> { resource_path: &'a str, name: &'a str, @@ -3057,33 +2852,6 @@ mod tests { assert_eq!(decode_ffi_error("ERROR not json"), "not json"); } - /// The engine says the same thing whether the cap came from the worker or is its own - /// default, so the attribution has to come from the config — naming the env var on a - /// worker that never set it would send the reader after the wrong problem. - #[test] - fn spill_cap_hint_attributes_the_limit_to_whoever_set_it() { - let capped = "Out of Memory Error: failed to offload data block of size 256.0 KiB \ - (18.9 MiB/19.0 MiB used).\nThis limit was set by the 'max_temp_directory_size' \ - setting.\nYou can adjust this setting, by using (for example) PRAGMA \ - max_temp_directory_size='10GiB'"; - - let configured = spill_cap_hint(capped, true).expect("the cap error must be recognised"); - assert!(configured.contains("DUCKDB_MAX_TEMP_DIRECTORY_SIZE")); - - let default = spill_cap_hint(capped, false).expect("the cap error must be recognised"); - assert!( - !default.contains("DUCKDB_MAX_TEMP_DIRECTORY_SIZE") && default.contains("low on disk"), - "with nothing configured the hint must point at the disk, not at an unset variable" - ); - - // Both must contradict the PRAGMA the engine suggests, which the lock always refuses. - assert!([configured, default] - .iter() - .all(|h| h.contains("PRAGMA suggested above is refused"))); - - assert!(spill_cap_hint("Binder Error: column \"x\" not found", true).is_none()); - } - fn test_ancestor(wid: &str) -> windmill_common::workspaces::DucklakeAncestorAttach { windmill_common::workspaces::DucklakeAncestorAttach { workspace_id: wid.to_string(), diff --git a/backend/windmill-worker/src/duckdb_runner.rs b/backend/windmill-worker/src/duckdb_runner.rs new file mode 100644 index 0000000000..46f5c5cbde --- /dev/null +++ b/backend/windmill-worker/src/duckdb_runner.rs @@ -0,0 +1,626 @@ +//! DuckDB jobs run in `windmill_duckdb_runner`, a process of their own, under the same job +//! isolation as every other language: inside nsjail when the worker jails jobs, under +//! `unshare` when it isolates that way, and as a plain child process otherwise. An engine +//! crash or a cancellation then ends the runner, never the worker. +//! +//! The runner hands back the string the engine call produced, so result collection and error +//! decoding stay in `duckdb_executor`. + +use std::{ + collections::HashSet, + path::{Path, PathBuf}, + process::Stdio, + sync::{LazyLock, Mutex}, + time::Duration, +}; + +use serde::Serialize; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWriteExt}, + process::Command, +}; +use windmill_common::error::{Error, Result}; + +use crate::{ + common::{build_command_with_isolation, resolve_nsjail_tmp_mount_block, DEV_CONF_NSJAIL}, + is_sandboxing_enabled, is_unshare_enabled, DISABLE_NUSER, HOME_ENV, NSJAIL_AVAILABLE, + NSJAIL_PATH, PATH_ENV, +}; + +const NSJAIL_CONFIG_RUN_DUCKDB_CONTENT: &str = include_str!("../nsjail/run.duckdb.config.proto"); + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub(crate) enum RunnerMode { + /// Inside nsjail: the jail confines the file system. + Jailed, + /// Under `unshare`, as other languages are on such a worker: its own PID namespace and a + /// cleared environment, with the worker user's view of the file system. + Unshare, + /// A plain child process, on a worker that does not isolate jobs. + Plain, +} + +/// Decided once per job, so every connection a job opens runs the same way. +pub(crate) fn runner_mode() -> RunnerMode { + if cfg!(unix) && is_sandboxing_enabled() && NSJAIL_AVAILABLE.is_some() { + RunnerMode::Jailed + } else if is_unshare_enabled() { + // `build_command_with_isolation` logs and runs unisolated when `unshare` is missing, + // as it does for every other language. + RunnerMode::Unshare + } else { + RunnerMode::Plain + } +} + +#[derive(Serialize, Clone, Copy)] +#[serde(rename_all = "snake_case")] +pub(crate) enum RequestMode { + Run, + Prepare, + Install, +} + +#[derive(Serialize, Clone, Copy)] +pub(crate) struct RunnerRequest<'a> { + pub mode: RequestMode, + pub blocks: &'a [String], + #[serde(skip_serializing_if = "Option::is_none")] + pub job_args: Option<&'a serde_json::value::RawValue>, + pub token: &'a str, + pub base_internal_url: &'a str, + pub w_id: &'a str, + pub memory_limit: &'a str, + pub temp_directory: &'a str, + pub max_result_size: usize, + pub collect_last_only: bool, + pub collect_first_row_only: bool, +} + +/// The runner's answer: the result string (`ERROR ` on failure) and the column order. +pub(crate) struct RunnerOutput { + pub result: String, + pub column_order: Option, +} + +fn runner_path() -> Result { + if let Ok(p) = std::env::var("DUCKDB_RUNNER_PATH") { + return Ok(p); + } + let exe = std::env::current_exe() + .map_err(|e| Error::InternalErr(format!("Could not locate the worker binary: {e}")))?; + let name = format!("windmill_duckdb_runner{}", std::env::consts::EXE_SUFFIX); + let path = exe.with_file_name(&name); + // Cargo runs test binaries from `target//deps`, one level below where the + // worker binary and `build_dev.sh`'s runner sit. + let in_target_dir = exe + .parent() + .filter(|d| d.ends_with("deps")) + .and_then(|d| d.parent()) + .map(|d| d.join(&name)); + let path = match in_target_dir { + Some(p) if !path.exists() && p.exists() => p, + _ => path, + }; + if !path.exists() { + return Err(Error::InternalErr(format!( + "DuckDB jobs need {}, alongside the worker binary. Build it with \ + windmill-duckdb-ffi-internal/build_dev.sh, or point DUCKDB_RUNNER_PATH at it.", + path.display() + ))); + } + Ok(path.to_string_lossy().into_owned()) +} + +/// The jail's own extension cache, separate from the worker user's `~/.duckdb` and mounted +/// read-only. Only [`install_missing_extensions`] writes to it, so a job can neither add to +/// nor replace what the next job loads from it. +fn extension_dir() -> String { + format!("{}/.duckdb/windmill_jail_extensions", HOME_ENV.as_str()) +} + +/// Extensions a job may load inside the jail, where the cache is read-only: the two +/// `setup_duckdb_connection` installs on every connection, the two the platform reaches for +/// without naming them (Parquet data files, the JSON type), and every extension a block +/// installs by name. A `FORCE INSTALL` stays forced: it is the one +/// statement here that must run every time. +fn needed_extensions(blocks: &[String]) -> Vec { + let mut out: Vec = ["httpfs", "azure", "parquet", "json"] + .iter() + .map(|s| format!("INSTALL {s};")) + .collect(); + for block in blocks { + let stmt = block.trim().trim_end_matches(';').trim(); + let mut toks = stmt.split_whitespace().peekable(); + let force = toks.peek().is_some_and(|t| t.eq_ignore_ascii_case("FORCE")); + if force { + toks.next(); + } + if !toks + .next() + .is_some_and(|t| t.eq_ignore_ascii_case("INSTALL")) + { + continue; + } + let Some(name) = toks.next() else { continue }; + if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') { + continue; + } + let name = name.to_ascii_lowercase(); + let force = if force { "FORCE " } else { "" }; + let stmt = match (toks.next(), toks.next(), toks.next()) { + (None, _, _) => format!("{force}INSTALL {name};"), + (Some(f), Some(r), None) + if f.eq_ignore_ascii_case("FROM") && r.eq_ignore_ascii_case("community") => + { + format!("{force}INSTALL {name} FROM community;") + } + _ => continue, + }; + if !out.contains(&stmt) { + out.push(stmt); + } + } + out +} + +static INSTALLED: LazyLock>> = LazyLock::new(Default::default); + +/// Installs, outside the jail, what the job will load inside it. The runner only runs +/// `INSTALL` here, and only of statements built above, so no script SQL and no extension +/// code runs unjailed. Once per extension per worker process. +async fn install_missing_extensions(blocks: &[String], job_dir: &str) -> Result<()> { + let missing: Vec = { + let installed = INSTALLED.lock().unwrap(); + needed_extensions(blocks) + .into_iter() + .filter(|s| s.starts_with("FORCE ") || !installed.contains(s)) + .collect() + }; + if missing.is_empty() { + return Ok(()); + } + let ext_dir = extension_dir(); + tokio::fs::create_dir_all(&ext_dir).await?; + let blocks: Vec = std::iter::once(format!( + "SET extension_directory='{}';", + ext_dir.replace('\'', "''") + )) + .chain(missing.iter().cloned()) + .collect(); + let out = spawn_runner( + RunnerMode::Plain, + &RunnerRequest { + mode: RequestMode::Install, + blocks: &blocks, + job_args: None, + token: "", + base_internal_url: "", + w_id: "", + memory_limit: "", + temp_directory: "", + max_result_size: 0, + collect_last_only: false, + collect_first_row_only: false, + }, + job_dir, + ) + .await?; + if let Some(raw) = out.result.strip_prefix("ERROR ") { + let msg = serde_json::from_str::(raw).unwrap_or_else(|_| raw.to_string()); + return Err(Error::ExecutionErr(format!( + "Installing DuckDB extensions for the jailed runner failed: {msg}" + ))); + } + INSTALLED.lock().unwrap().extend(missing.into_iter().filter(|s| !s.starts_with("FORCE "))); + Ok(()) +} + +/// `FORCE INSTALL` rewrites the cache, which is read-only inside the jail. The worker has +/// already honored it outside, so inside it becomes a plain `INSTALL`, which finds the +/// extension present and does nothing. Same block count, so result collection is unchanged. +fn unforce_installs(blocks: &[String]) -> Vec { + blocks + .iter() + .map(|b| { + let t = b.trim_start(); + match t.split_once(char::is_whitespace) { + Some((first, rest)) + if first.eq_ignore_ascii_case("FORCE") + && rest + .trim_start() + .get(..7) + .is_some_and(|k| k.eq_ignore_ascii_case("INSTALL")) => + { + rest.trim_start().to_string() + } + _ => b.clone(), + } + }) + .collect() +} + +/// The only place a jailed runner can write on the host: its spill directory and its result +/// file. The rest of the job dir is mounted read-only. +fn work_dir(job_dir: &str) -> String { + format!("{job_dir}/duckdb") +} + +/// DuckDB's spill cap, in DuckDB's spelling. The cloud runs one worker per pod on nodes with no +/// per-pod ephemeral-storage limit, so it gets one even when nothing configured it. +fn max_temp_directory_size() -> Option { + match std::env::var("DUCKDB_MAX_TEMP_DIRECTORY_SIZE") { + Ok(v) if !v.trim().is_empty() => Some(v.trim().to_string()), + _ if *windmill_common::worker::CLOUD_HOSTED => Some("6GiB".to_string()), + _ => None, + } +} + +/// Parses a DuckDB size (`6GiB`, `500 MB`, `1024`): 1024-based for `KiB`..`TiB`, 1000-based +/// for `KB`..`TB`. `None` for anything else, which DuckDB then refuses on its own. +fn parse_duckdb_size(s: &str) -> Option { + let s = s.trim(); + let split = s + .find(|c: char| !(c.is_ascii_digit() || c == '.')) + .unwrap_or(s.len()); + let (num, unit) = s.split_at(split); + let num: f64 = num.parse().ok()?; + let mult: f64 = match unit.trim().to_ascii_lowercase().as_str() { + "" | "b" | "byte" | "bytes" => 1.0, + "kb" => 1e3, + "mb" => 1e6, + "gb" => 1e9, + "tb" => 1e12, + "kib" => 1024.0, + "mib" => 1024.0 * 1024.0, + "gib" => 1024.0 * 1024.0 * 1024.0, + "tib" => 1024.0 * 1024.0 * 1024.0 * 1024.0, + _ => return None, + }; + Some((num * mult) as u64) +} + +pub(crate) async fn run( + mode: RunnerMode, + req: &RunnerRequest<'_>, + job_dir: &str, +) -> Result { + let spill_dir = format!("{}/spill", work_dir(job_dir)); + tokio::fs::create_dir_all(&spill_dir).await?; + let req = RunnerRequest { temp_directory: &spill_dir, ..*req }; + if mode != RunnerMode::Jailed { + return spawn_runner(mode, &req, job_dir).await; + } + install_missing_extensions(req.blocks, job_dir).await?; + let blocks = unforce_installs(req.blocks); + spawn_runner(mode, &RunnerRequest { blocks: &blocks, ..req }, job_dir).await +} + +/// Disk the directory actually uses (allocated blocks, so sparse files count for what they +/// hold), without following symlinks: the jailed script controls what is in there. +fn disk_usage(dir: &Path) -> u64 { + let mut total = 0; + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&d) else { + continue; + }; + for entry in entries.flatten() { + let Ok(meta) = entry.path().symlink_metadata() else { + continue; + }; + #[cfg(unix)] + { + total += std::os::unix::fs::MetadataExt::blocks(&meta) * 512; + } + #[cfg(not(unix))] + { + total += meta.len(); + } + if meta.is_dir() { + stack.push(entry.path()); + } + } + } + total +} + +/// Resolves once the runner's work dir holds more than `cap`. Inside the jail that dir is the +/// only place the script can write to on the host, and the script can raise DuckDB's own +/// `max_temp_directory_size`, so this is what makes the cap hold. +async fn exceeds(dir: PathBuf, cap: u64) -> u64 { + loop { + tokio::time::sleep(Duration::from_secs(1)).await; + let d = dir.clone(); + let used = tokio::task::spawn_blocking(move || disk_usage(&d)) + .await + .unwrap_or(0); + if used > cap { + return used; + } + } +} + +/// Kills the runner's process group when the job future is dropped (cancellation, timeout) +/// before the runner exits. Killing the spawned process alone is not enough: `unshare --fork` +/// leaves its child running. +struct KillGroupOnDrop(Option); + +impl Drop for KillGroupOnDrop { + fn drop(&mut self) { + #[cfg(unix)] + if let Some(pgid) = self.0 { + let _ = nix::sys::signal::killpg( + nix::unistd::Pid::from_raw(pgid), + nix::sys::signal::Signal::SIGKILL, + ); + } + } +} + +async fn spawn_runner( + mode: RunnerMode, + req: &RunnerRequest<'_>, + job_dir: &str, +) -> Result { + let runner = runner_path()?; + let call_id = uuid::Uuid::new_v4(); + let work_dir = work_dir(job_dir); + tokio::fs::create_dir_all(&work_dir).await?; + let result_path = format!("{work_dir}/result_{call_id}.out"); + let spill_cap = max_temp_directory_size(); + + let mut cmd = match mode { + RunnerMode::Jailed => { + let ext_dir = extension_dir(); + tokio::fs::create_dir_all(&ext_dir).await?; + let bigquery_mount = match std::env::var("GOOGLE_APPLICATION_CREDENTIALS") { + Ok(p) if !p.is_empty() => format!( + "mount {{\n src: \"{p}\"\n dst: \"{p}\"\n is_bind: true\n mandatory: false\n}}" + ), + _ => String::new(), + }; + let config_name = format!("run.duckdb.{call_id}.config.proto"); + tokio::fs::write( + format!("{job_dir}/{config_name}"), + NSJAIL_CONFIG_RUN_DUCKDB_CONTENT + .replace("{JOB_DIR}", job_dir) + .replace("{RUNNER_PATH}", &runner) + .replace("{EXTENSION_DIR}", &ext_dir) + .replace("{BIGQUERY_CREDENTIALS_MOUNT}", &bigquery_mount) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) + .replace("#{DEV}", DEV_CONF_NSJAIL) + // Rooted at the work dir rather than the job dir: with a disk-backed /tmp + // the jail's /tmp is a host directory, and inside the work dir the disk cap + // covers it. A RAM-backed /tmp is bounded by its tmpfs size. + .replace( + "{TMP_MOUNT_BLOCK}", + &resolve_nsjail_tmp_mount_block(&work_dir).await, + ), + ) + .await?; + let mut cmd = Command::new(NSJAIL_PATH.as_str()); + cmd.args([ + "--config", + &config_name, + "--", + "/tmp/windmill_duckdb_runner", + ]); + cmd + } + RunnerMode::Unshare => build_command_with_isolation(&runner, &[]), + RunnerMode::Plain => Command::new(&runner), + }; + + cmd.current_dir(job_dir) + .env_clear() + .env("PATH", PATH_ENV.as_str()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + #[cfg(unix)] + cmd.process_group(0); + // The jail sets its own HOME, over the read-only extension cache. + if mode != RunnerMode::Jailed { + cmd.env("HOME", HOME_ENV.as_str()); + } + // Windows networking (Winsock) needs SystemRoot, so without these every extension install + // and S3 proxy request fails. Empty off Windows. + cmd.envs(crate::WIN_ENVS.iter().cloned()); + #[cfg(windows)] + cmd.env("windir", crate::SYSTEM_ROOT.as_str()); + // DuckDB's default TimeZone follows TZ, as it did in the worker's own environment. + if !crate::TZ_ENV.is_empty() { + cmd.env("TZ", crate::TZ_ENV.as_str()); + } + if let Some(cap) = &spill_cap { + cmd.env("DUCKDB_MAX_TEMP_DIRECTORY_SIZE", cap); + } + for var in [ + "GOOGLE_APPLICATION_CREDENTIALS", + "HTTP_PROXY", + "HTTPS_PROXY", + "NO_PROXY", + "ALL_PROXY", + "http_proxy", + "https_proxy", + "no_proxy", + "all_proxy", + ] { + if let Ok(v) = std::env::var(var) { + cmd.env(var, v); + } + } + + let mut payload = serde_json::to_value(req).map_err(|e| Error::InternalErr(e.to_string()))?; + payload["result_path"] = serde_json::Value::String(result_path.clone()); + let payload = serde_json::to_vec(&payload).map_err(|e| Error::InternalErr(e.to_string()))?; + + // The engine's memory lives outside the worker, so a native OOM can take the job alone, as + // long as the kernel picks the runner first. Set before exec so nsjail or unshare, and the + // runner they start, inherit it: written after spawn it races that, and the jail's /proc is + // read-only. Only raw syscalls here, between fork and exec. + #[cfg(target_os = "linux")] + { + let adj = windmill_common::worker::JOB_OOM_SCORE_ADJ.to_string(); + unsafe { + cmd.pre_exec(move || { + let fd = nix::libc::open(c"/proc/self/oom_score_adj".as_ptr(), nix::libc::O_WRONLY); + if fd >= 0 { + nix::libc::write(fd, adj.as_ptr().cast(), adj.len()); + nix::libc::close(fd); + } + Ok(()) + }); + } + } + let mut child = cmd.spawn().map_err(|e| { + Error::InternalErr(format!("Could not start the DuckDB runner ({mode:?}): {e}")) + })?; + let mut guard = KillGroupOnDrop(child.id().map(|p| p as i32)); + let mut stdin = child.stdin.take().expect("stdin is piped"); + let write = async move { + stdin.write_all(&payload).await?; + stdin.shutdown().await + }; + let stdout = child.stdout.take().expect("stdout is piped"); + let stderr = child.stderr.take().expect("stderr is piped"); + let cap_bytes = spill_cap.as_deref().and_then(parse_duckdb_size); + let watchdog = async { + match cap_bytes { + Some(cap) if mode == RunnerMode::Jailed => exceeds(PathBuf::from(&work_dir), cap).await, + _ => std::future::pending().await, + } + }; + let (write, status, stdout, stderr) = tokio::select! { + r = async { tokio::join!(write, child.wait(), output_tail(stdout), output_tail(stderr)) } => r, + // Returning drops the child, and with it `kill_on_drop` and `guard` end the runner. + used = watchdog => { + return Err(Error::ExecutionErr(format!( + "DuckDB: this job wrote {} MB to disk, past this worker's limit of {} \ + (DUCKDB_MAX_TEMP_DIRECTORY_SIZE), and was stopped. Raising \ + max_temp_directory_size from the script does not lift it. Spill less (fewer \ + threads, a narrower scan, aggregating before the join), or write large outputs \ + to S3 rather than local files.", + used / 1_000_000, + spill_cap.as_deref().unwrap_or_default() + ))); + } + }; + let status = status?; + guard.0 = None; + let read = read_result(&result_path).await; + let _ = tokio::fs::remove_file(&result_path).await; + + if !status.success() || read.is_err() { + #[cfg(unix)] + let signal = std::os::unix::process::ExitStatusExt::signal(&status); + #[cfg(not(unix))] + let signal: Option = None; + let how = match signal { + Some(sig) => format!("was killed by signal {sig}"), + None => format!("exited with {status}"), + }; + return Err(Error::ExecutionErr(format!( + "The DuckDB engine process {how}. The worker is unaffected.{}{}{}", + write + .err() + .map(|e| format!("\nrequest: {e}")) + .unwrap_or_default(), + if stderr.trim().is_empty() { + String::new() + } else { + format!("\nstderr: {}", stderr.trim()) + }, + if stdout.trim().is_empty() { + String::new() + } else { + format!("\nstdout: {}", stdout.trim()) + }, + ))); + } + let mut column_order = read?; + let header_end = column_order.find('\n').unwrap_or(column_order.len()); + let result = column_order.split_off((header_end + 1).min(column_order.len())); + column_order.truncate(header_end); + Ok(RunnerOutput { column_order: (!column_order.is_empty()).then_some(column_order), result }) +} + +/// The result file sits in a directory the jailed script can write to, so a symlink planted there +/// must not make the worker read some other host file into the job result. +async fn read_result(path: &str) -> std::io::Result { + let mut opts = tokio::fs::OpenOptions::new(); + opts.read(true); + #[cfg(unix)] + opts.custom_flags(nix::libc::O_NOFOLLOW); + let mut s = String::new(); + opts.open(path).await?.read_to_string(&mut s).await?; + Ok(s) +} + +/// Drains a runner pipe keeping only its last few KB. A script controls how much the engine +/// prints (`CALL enable_logging(storage='stdout')`), so buffering it whole would let a jailed +/// job grow the worker's memory without bound. +async fn output_tail(mut r: impl AsyncRead + Unpin) -> String { + const KEEP: usize = 4096; + let mut buf = vec![0u8; 16 * 1024]; + let mut tail: Vec = Vec::with_capacity(2 * KEEP); + loop { + match r.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => { + tail.extend_from_slice(&buf[..n]); + if tail.len() > KEEP { + tail.drain(..tail.len() - KEEP); + } + } + } + } + String::from_utf8_lossy(&tail).into_owned() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn duckdb_sizes_parse_in_duckdb_units() { + assert_eq!(parse_duckdb_size("6GiB"), Some(6 * 1024 * 1024 * 1024)); + assert_eq!(parse_duckdb_size("500 MB"), Some(500_000_000)); + assert_eq!(parse_duckdb_size("1.5kib"), Some(1536)); + assert_eq!(parse_duckdb_size("1024"), Some(1024)); + // Kubernetes' spelling is not DuckDB's: the engine refuses it, so no cap is derived. + assert_eq!(parse_duckdb_size("6Gi"), None); + assert_eq!(parse_duckdb_size(""), None); + } + + fn strs(v: &[&str]) -> Vec { + v.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn force_installs_run_outside_and_become_plain_inside() { + let blocks = strs(&[ + "force install httpfs;", + "INSTALL bigquery FROM community;", + "INSTALL x FROM 'http://evil';", + "INSTALL '/tmp/x.so';", + "SELECT 'FORCE INSTALL';", + ]); + assert_eq!( + needed_extensions(&blocks)[4..], + strs(&["FORCE INSTALL httpfs;", "INSTALL bigquery FROM community;"]) + ); + assert_eq!( + unforce_installs(&blocks), + strs(&[ + "install httpfs;", + "INSTALL bigquery FROM community;", + "INSTALL x FROM 'http://evil';", + "INSTALL '/tmp/x.so';", + "SELECT 'FORCE INSTALL';", + ]) + ); + } +} diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index aa1d9ab209..e56334df67 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -45,8 +45,8 @@ mod deno_executor; mod docker_v2; #[cfg(feature = "duckdb")] mod duckdb_executor; -#[cfg(all(feature = "duckdb", feature = "private"))] -mod duckdb_isolation_ee; +#[cfg(feature = "duckdb")] +mod duckdb_runner; mod global_cache; mod go_executor; mod graphql_executor; diff --git a/docker/RHEL8/Dockerfile b/docker/RHEL8/Dockerfile index ea53feabf1..a7eca5fbd8 100644 --- a/docker/RHEL8/Dockerfile +++ b/docker/RHEL8/Dockerfile @@ -93,7 +93,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ CARGO_NET_GIT_FETCH_WITH_CLI=true cargo build --release RUN mkdir -p /usr/src/app && \ - cp windmill-duckdb-ffi-internal/target/release/libwindmill_duckdb_ffi_internal.so /usr/src/app/ + cp windmill-duckdb-ffi-internal/target/release/windmill_duckdb_runner /usr/src/app/ # Runtime Kerberos packages (installed separately from build deps) RUN yum install -y krb5-workstation cyrus-sasl-gssapi diff --git a/docker/RHEL9/Dockerfile b/docker/RHEL9/Dockerfile index ab94df5fc4..cc972884d8 100644 --- a/docker/RHEL9/Dockerfile +++ b/docker/RHEL9/Dockerfile @@ -93,7 +93,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ CARGO_NET_GIT_FETCH_WITH_CLI=true cargo build --release RUN mkdir -p /usr/src/app && \ - cp windmill-duckdb-ffi-internal/target/release/libwindmill_duckdb_ffi_internal.so /usr/src/app/ + cp windmill-duckdb-ffi-internal/target/release/windmill_duckdb_runner /usr/src/app/ # Runtime Kerberos packages (installed separately from build deps) RUN yum install -y krb5-workstation cyrus-sasl-gssapi diff --git a/docs/duckdb-execution.md b/docs/duckdb-execution.md new file mode 100644 index 0000000000..70f9218ee4 --- /dev/null +++ b/docs/duckdb-execution.md @@ -0,0 +1,118 @@ +# How DuckDB jobs run + +A DuckDB job runs in `windmill_duckdb_runner`, a binary built from +`backend/windmill-duckdb-ffi-internal` (`src/bin`) and shipped next to the `windmill` binary. +The worker spawns one runner per DuckDB connection a job opens: the job itself, and the +`on_schema_change=sync` pre-pass. It never loads the engine into its own process. + +The runner gets the same job isolation as every other language. `duckdb_runner::runner_mode` +decides it per job: + +| worker isolates with | runner runs | what it can see | +|---|---|---| +| nsjail | inside `run.duckdb.config.proto` | read-only `/usr`, `/lib`, `/etc`; its own `/tmp`; its job dir read-only, except `duckdb/` (spill + result); a read-only extension cache | +| unshare | under `build_command_with_isolation` | the worker user's file system, its own PID namespace, a cleared environment | +| nothing | as a plain child process | the worker user's file system, a cleared environment | + +There is no SQL-level policy on top. A script can read and write what the jail lets it and load +any extension DuckDB will sign, as a Python job can read the same files and load native code. +Under nsjail its disk use is capped from the worker (below). Network egress is open in every +mode, as it is for the other languages: the S3 proxy, metadata endpoints and internal hosts are +all reachable. + +## What crosses to the runner + +`spawn_runner` writes one JSON request to the runner's stdin, so nothing secret is put on argv or +into the environment: + +- the blocks, already rewritten by the worker: resource `ATTACH`es, DuckLake and datatable + attaches with their credentials inlined, workspace macros; +- the job args, the job token, `base_internal_url` and the workspace id. The runner derives the + S3/GCS secrets for the workspace's S3 proxy from the token, which it reaches over the network; +- `memory_limit` (`DUCKDB_MEMORY_LIMIT`, or 80% of the worker's cgroup), the temp directory (the + job dir), and the result-size cap (`MAX_SQL_RESULT_SIZE`). + +The environment is cleared except `PATH`, `HOME` (outside nsjail), `DUCKDB_MAX_TEMP_DIRECTORY_SIZE`, +`GOOGLE_APPLICATION_CREDENTIALS` (BigQuery `ATTACH`, whose credentials file nsjail mounts) and the +proxy variables. In particular a script cannot use the worker's own `AWS_*` credential chain. + +Under nsjail the job dir is mounted read-only at its host path, so every path the worker wrote +into the blocks (Postgres root certificates) resolves unchanged. The one writable place on the +host is `{job dir}/duckdb/`: the spill directory and the result file, which the worker opens +without following symlinks, since the script can write there. + +The runner writes its answer to a file in that directory: a first line with the last statement's +column order, then either the collected results or `ERROR `. +`finish_run` / `finish_prepare` in `duckdb_executor.rs` turn that into the job result. + +## Failure, cancellation and memory + +- **A crash fails the job, never the worker.** A runner that exits without writing its result + becomes `The DuckDB engine process … The worker is unaffected.` with the tail of its stderr. + An engine `std::terminate` (for example a remote `temp_directory` that has to spill) and a + panic in result rendering both land here. +- **Cancellation and timeouts kill the engine.** The executor drops the job future, and + `KillGroupOnDrop` kills the runner's whole process group. Killing the spawned process alone is + not enough: `unshare --fork` leaves its child running. +- **The kernel OOM-kills the runner, not the worker.** `JOB_OOM_SCORE_ADJ` is written in + `pre_exec`, before nsjail or `unshare` is exec'd, so the runner inherits it. Written after spawn + it would race their fork, and the jail's `/proc` is read-only. Within the job, DuckDB bounds + itself through `memory_limit` and spills past it. +- **Runner output is bounded.** stdout/stderr are drained keeping only the last 4 KB: a script + controls how much the engine prints (`CALL enable_logging(storage='stdout')`). + +## Extensions + +Outside nsjail the runner uses the worker user's `~/.duckdb/extensions` and installs whatever a +script names. + +Under nsjail it gets its own cache, `~/.duckdb/windmill_jail_extensions`, mounted read-only. +Only the worker writes to it: before a jailed run it starts the runner unjailed in `install` +mode, with only `INSTALL` statements it built itself, for `httpfs`, `azure`, `parquet`, `json` +and every extension a block installs by name. That runs no script SQL and loads nothing, so no +extension code runs outside the jail. It happens once per extension per worker process, except +for `FORCE INSTALL`, which is honored every time and becomes a plain `INSTALL` inside the jail. +A job can therefore neither add to nor replace what the next job loads from that cache. An +extension used without being named and outside those four (`icu` behind `SET TimeZone`, say) +fails on the read-only mount: name it with `INSTALL`. + +A script can still point `extension_directory` somewhere writable and install into it. That only +affects its own connection, and is no more than a Python job could do. + +## The engine + +The runner links duckdb-rs from crates.io, pinned with `=` so `cargo update` cannot move the +engine. The engine loads the **prebuilt** extensions from `extensions.duckdb.org`, compiled for the +exact release it reports, so an engine bump is a version change in `Cargo.toml` and nothing else. + +Confinement belongs to the job isolation, not to engine settings. DuckDB spills through the same +file system a query reads with, so `disabled_filesystems='LocalFileSystem'` also stops every query +that outgrows `memory_limit`, and no combination of stock settings gives "spill, but no other local +file access". Fencing the engine at the SQL level again would need a patched engine to keep +spilling. + +## Bounding disk use + +The cap is `DUCKDB_MAX_TEMP_DIRECTORY_SIZE`, or `6GiB` on a `CLOUD_HOSTED` worker that does not +set it. The cloud runs one worker per pod on nodes with no per-pod ephemeral-storage limit, and a +worker pod already holds several GiB of runtime caches, so the cap is sized against a mostly full +node; re-measure before raising it. Unset elsewhere, DuckDB's own default applies: 90% of the +temp volume's free space. + +It is applied twice: + +- **DuckDB's `max_temp_directory_size`**, set on every connection, so an ordinary query that + outgrows it fails with the engine's `failed to offload data block … max_temp_directory_size` + error and cleans up its temp files. +- **A worker-side watchdog**, under nsjail. A script can raise the engine setting, so on its own + it is only a default. The worker sums the allocated size of `{job dir}/duckdb/` once a second + and kills the runner when it passes the cap, whatever wrote the bytes: spill, `COPY … TO`, or + the result file. With a disk-backed jail `/tmp` (`NSJAIL_TMP_BACKING=disk`) that `/tmp` lives +inside the same directory, so it counts too; a RAM-backed one is bounded by its tmpfs size. +Enforcement lags by up to a second of writing. Outside nsjail the script can + write anywhere the worker user can, as other languages can, so there the cap stays a default. + +The value is DuckDB's spelling, not Kubernetes': `KiB`/`MiB`/`GiB`/`TiB` for 1024-based units, +`KB`/`MB`/`GB`/`TB` for 1000-based. An unparseable value fails every DuckDB job on the worker +rather than being ignored. `DUCKDB_MEMORY_LIMIT` is spelled the same way, and the two are set in +separate statements so a parse error names the right variable. diff --git a/docs/duckdb-isolation.md b/docs/duckdb-isolation.md deleted file mode 100644 index fb85085b2c..0000000000 --- a/docs/duckdb-isolation.md +++ /dev/null @@ -1,223 +0,0 @@ -# DuckDB isolation and the patched engine - -DuckDB runs **in-process** in the worker, through the FFI cdylib in -`backend/windmill-duckdb-ffi-internal`. It is therefore the one language that the -nsjail/unshare isolation cannot confine: there is no child process to jail. On a worker -that isolates jobs, `windmill-worker/src/duckdb_isolation_ee.rs` mirrors that policy at -the connection level instead, rewriting the block list handed to the engine so it starts -with three irreversible settings: - -```sql -SET lock_temp_directory=true; -SET disabled_filesystems='LocalFileSystem'; -SET allow_community_extensions=false; -``` - -`disabled_filesystems` is the fence. `lock_temp_directory` is what keeps out-of-core -execution working behind it, and it is **not a stock DuckDB setting** — see below. - -## Why the engine had to change - -DuckDB spills through the same file system a query would use to read `/etc/hostname`, so -`disabled_filesystems='LocalFileSystem'` also stops the buffer manager. Every query whose -working set outgrows `memory_limit` then fails instead of going out-of-core. No -combination of stock settings gives "spill, but no other local file access": - -- `allowed_directories` / `allowed_paths` enforce nothing while `enable_external_access` - is true — `DBConfig::CanAccessFile` returns early. -- Turning `enable_external_access` off does preserve spilling (startup adds the temp - directory to `allowed_directories`), but the postgres and mysql extensions refuse to - `ATTACH` on that flag alone, with no allowlist entry able to re-permit it. That would - take DuckLake catalogs and datatables down with it. -- A remote `temp_directory` is impossible: DuckDB cannot open an HTTP file for both - reading and writing, and the attempt aborts the process. - -The engine is also not arranged the way the layering suggests. `FileSystem::GetLocal(db)` -looks like an escape hatch — it is what secret storage and extension installation use — -but `LocalDatabaseFileSystem::GetFileSystem()` re-applies the `disabled_filesystems` -check by hand, so it is not one. (That is also why extension autoload broke behind the -fence in DuckDB 1.5.3, which `EXTENSION_ALLOWLIST` and `PRELOAD_EXTENSIONS` exist to work -around.) The buffer manager, meanwhile, reaches its temp files through -`FileSystem::GetFileSystem(db)`, an opener-wrapped virtual file system, so both the path -allowlist and the disable apply to it. - -## What the patches add - -`lock_temp_directory` (BOOLEAN, default false, global, irreversible). Setting it: - -1. fixes `temp_directory` in place — it can no longer be modified, and the setting itself - cannot be unset or reset; -2. fixes `max_temp_directory_size` the same way, so the ceiling on what may be written - there cannot be raised either; -3. routes the buffer manager's temporary files through a local file system that - `disabled_filesystems` does not apply to. - -The halves are what make it safe as a set. The exemption is only sound because the -directory can no longer be redirected: without (1), a script could point `temp_directory` -at an arbitrary path and get a blind write primitive through the exemption. And it is only -bounded because of (2): DuckDB accepts `SET max_temp_directory_size='100GB'` from any -statement, so a cap a script can lift is not a cap. -`allowed_directories` still applies either way. - -The isolation transform therefore emits `SET lock_temp_directory=true` **before** -`disabled_filesystems`, while the local file system is still reachable, and after the FFI -has already pointed `temp_directory` at the job dir -(`configure_duckdb_resource_limits`). Everything a script authored runs after all three -settings, so a script can neither move the temp directory nor raise the cap on it. - -## Bounding the spill - -`configure_duckdb_resource_limits` sets `max_temp_directory_size` from -`DUCKDB_MAX_TEMP_DIRECTORY_SIZE`, immediately before it takes the lock. Unset — the -self-hosted default — leaves DuckDB's own default of 90% of the temp volume's free space, -so nothing changes for an install that does not opt in. The cloud sets `6GiB` per worker -pod (`cloudee/cloud/cloud.yml` in `infra-aws`); those pods run a single worker each, so it -is effectively a per-pod cap and is not divided by `NUM_WORKERS`. - -Those pods carry no `ephemeral-storage` limit, so what the cap has to stay clear of is the -node's eviction threshold rather than a per-pod one, and the budget is tighter than the -node's size suggests: a worker pod already holds 4.7-6.9 GiB of language-runtime caches and -job dirs before any spill, so a ~100 GB node running 5-6 of them sits at 45-60 GB used at -rest. Every pod on the fullest node spilling to its cap at the same instant would land near -the kubelet's default `nodefs.available<10%`. That is the tail, not the common case, and it -is still bounded where the previous behaviour was not — one query could take all the free -space by itself — but the cap is sized against a mostly-full node, not an empty one. Re-measure -before raising it, and re-measure if the runtime-cache baseline grows. - -It is read from the environment rather than passed across the FFI: the cdylib is versioned -separately from the worker and agent workers ship it out of band, so a signature change -would mean an ABI bump and a coordinated redeploy for what is a config value. - -The value is DuckDB's spelling, not Kubernetes': `KiB`/`MiB`/`GiB`/`TiB` for 1024-based -units, `KB`/`MB`/`GB`/`TB` for 1000-based. Kubernetes' `6Gi` is not a unit DuckDB knows, -and an unparseable value fails every DuckDB job on the worker rather than being ignored — -a cap that quietly disappeared would be worse. `DUCKDB_MEMORY_LIMIT` is spelled the same -way and DuckDB's parse error says only "memory", so the cap's `SET` is issued on its own -statement: batched together, a bad value in either would produce the same message and name -neither. - -Enforcement is graceful. A query that needs more spill than the cap fails with -`Out of Memory Error: failed to offload data block of size … This limit was set by the -'max_temp_directory_size' setting.`, its temp files are removed, and the worker carries on. -Size it from measurement, not from the data: DuckDB compresses temp files, so the logical -spill a given cap permits is well above the number set. - -That engine message goes on to suggest `PRAGMA max_temp_directory_size='10GiB'`, which the -lock refuses — advice that was actionable before the cap was frozen and is now a dead end -ending in a second, unrelated-looking permission error. `decode_ffi_error` -(`duckdb_executor.rs`) appends a correction whenever the message appears, on every worker -rather than only isolating ones, since the lock is taken on every connection. It matches -DuckDB's wording verbatim, which -`spilling_past_the_configured_cap_fails_the_query_and_cleans_up` asserts against real engine -output so an engine bump that rewords it fails the test instead of silently dropping the -correction. - -The engine emits that same message for **its own default cap**, so which limit was hit -cannot be read off the message: with `DUCKDB_MAX_TEMP_DIRECTORY_SIZE` unset it means the -worker is low on disk, not that anyone configured a ceiling. `spill_cap_hint` takes the -answer from the worker's config instead and says whichever is true — naming the env var on a -worker that never set it would point the reader at a knob that does not exist. - -## Where the patched engine comes from - -`backend/windmill-duckdb-ffi-internal` depends on -[`windmill-labs/duckdb-rs`](https://github.com/windmill-labs/duckdb-rs), branch -`windmill/duckdb-1.5.5`, pinned by rev. That fork is duckdb-rs 1.10505.0 (DuckDB 1.5.5) -plus the windmill commits; `duckdb.tar.gz` is left byte-identical to upstream's and -`crates/libduckdb-sys/windmill/0*.patch` are applied to the extracted sources at build -time, in order, so `git diff` against `duckdb/duckdb-rs` shows the whole delta as text. -`0001` adds the setting and the exemption, `0002` extends the lock to -`max_temp_directory_size`. `crates/libduckdb-sys/windmill/README.md` in the fork carries -the rebase recipe. - -### The patches may not move a struct member - -The engine loads the **prebuilt** extensions from `extensions.duckdb.org` — httpfs, azure, -postgres, ducklake, parquet, and the rest of `EXTENSION_ALLOWLIST` — which are compiled -against the struct layouts of the release the engine claims to be. A patch that shifts a -member those extensions reach breaks them, and not loudly: the extension reads the -neighbouring field and `LOAD httpfs` **blocks forever** rather than failing, so every -DuckDB job hangs until its timeout. - -The first revision of `0001` declared its new `DatabaseInstance` member next to -`local_db_file_system`, which pushed `create_api_v1` — the extension C-API entry point — -8 bytes along, and every extension load hung. It is declared last for that reason, and -`lock_temp_directory` is declared at the end of `DBConfigOptions`' run of bools so it takes -a byte of padding the struct already had. - -The invariant is **no pre-existing member changes offset**, which is strictly stronger than -`sizeof` holding — the first attempt at that field went in next to `temporary_directory`, -kept `sizeof(DBConfigOptions)` at 920 by consuming interior padding, and still slid four -bools along by a byte each. A rebase must diff the full record layouts -(`clang++ -Xclang -fdump-record-layouts`, recipe in the fork's README), not the sizes. -`prebuilt_extensions_still_load` in the FFI crate is the end-to-end guard — a timeout, not -an assertion, because the failure mode is a hang — but it covers extension *entry*, not -every field an extension reads, so it does not substitute for the layout diff. - -Alternatives considered and rejected: vendoring a patched `libduckdb-sys` in this repo -would put a 6 MB binary tarball into windmill's history on every engine bump and would -need the FFI builder stage in `Dockerfile` to copy more than -`backend/windmill-duckdb-ffi-internal`; linking a prebuilt engine through -`DUCKDB_LIB_DIR` would drop the `bundled` build, so every developer and every CI job -would need a patched `libduckdb` on the box before `build_dev.sh` could work. - -## What stops the patches disappearing silently - -An unpatched engine compiles, links and passes every existing test. Three things catch it: - -- **The fork's build script** asserts each patch's marker string is present in the patched - sources after applying it, and panics if one no longer applies. -- **`patched_engine_tests` in the FFI crate** runs a query that must spill, first - without the lock (asserting the fence stops it, which is what proves the query really - spills) and then with it (asserting it succeeds); it also pins that the lock leaves - neither `temp_directory` nor `max_temp_directory_size` settable, and that a configured - connection stops at its cap. It fails outright on a stock engine, where the `SET` is an - unrecognized parameter. CI runs it via the FFI crate's own `cargo test` step in - `backend-test.yml`. -- **The transform itself fails closed**: on a stock engine `SET lock_temp_directory=true` - errors, so every DuckDB job on an isolating worker fails immediately rather than - quietly losing the ability to spill. - -## Upstream - -Both patches are written against `duckdb/duckdb` `v1.5.5` in the shape upstream takes -changes. Besides the engine sources they carry `src/common/settings.json` (the settings -generator's input, regenerated with `scripts/generate_settings.py` and formatted with -clang-format 11.0.1), two sqllogictests under `test/sql/settings/`, and one line in -`test/api/test_reset.cpp` — that test requires every setting to survive a `RESET`, so an -irreversible one has to join the exclusion list `disabled_filesystems` and -`lock_configuration` are already on. They are `git format-patch` files, so they apply with -`git am`: - -```bash -git clone https://github.com/duckdb/duckdb.git && cd duckdb && git checkout v1.5.5 -git am .../crates/libduckdb-sys/windmill/0*.patch -GEN=ninja make && ./build/release/test/unittest "test/sql/settings/lock_temp_directory*" -``` - -`0001` is filed as -**[duckdb/duckdb#24694](https://github.com/duckdb/duckdb/pull/24694)**, against `main` -rather than `v1.5.5`. Filing was a human decision on purpose: their `CONTRIBUTING.md` asks -contributors not to submit LLM-generated pull requests, to discuss the change on GitHub -first, and to run CI on a fork before opening one. - -**`0002` is windmill-only and is not on #24694.** It is kept as a separate commit rather -than folded into `0001` for exactly that reason: `0001` stays byte-identical to what the -open PR carries, and `0002` applies to it verbatim as a second commit whenever a human -chooses to push it. That is the one sanctioned divergence; when it is pushed, say so here -and in the fork's `windmill/README.md`, and the "same change" rule below covers it too. - -**[duckdb/duckdb#24695](https://github.com/duckdb/duckdb/issues/24695)** states the -underlying problem without prescribing a fix, and is where a different design would land: -`lock_temp_directory` is one shape, and an enforceable `allowed_directories` would solve it -equally well. Watch the issue rather than the PR for whether this fork can be retired. - -`0001` and #24694 are the same change and must stay that way. They differ only -where the base version forces it: the `DUCKDB_SETTING_ALIAS` indices, `vfs` versus -`db.config.file_system` in `FileSystem::GetLocal`, and the sqllogictest directory -placeholder. Nothing else should diverge. Note that the build applies neither the `test/` -hunks nor `settings.json` (`NOT_IN_TARBALL` in `build_windmill_patch.rs`), so drift in -those is invisible here and only surfaces when #24694 is rebased. - -Until it lands in a released DuckDB the fork is the delivery mechanism, and the exit path -is to delete `crates/libduckdb-sys/windmill/` and go back to the crates.io crate. diff --git a/docs/external-instance-datatables.md b/docs/external-instance-datatables.md index acfb93b90a..38e810b03d 100644 --- a/docs/external-instance-datatables.md +++ b/docs/external-instance-datatables.md @@ -83,6 +83,6 @@ Worth knowing while testing: - A worker needs the `postgresql` and `duckdb` tags for those jobs (`update config set config = jsonb_set(config, '{worker_tags}', …) where name = 'worker__default'`). -- DuckDB jobs load `libwindmill_duckdb_ffi_internal.so` by name, so a binary built into its own - `CARGO_TARGET_DIR` needs that library on `LD_LIBRARY_PATH`. +- DuckDB jobs run `windmill_duckdb_runner` from next to the worker binary, so a binary built + into its own `CARGO_TARGET_DIR` needs `DUCKDB_RUNNER_PATH` pointed at the runner. - Setup holds the lifecycle lock for its whole run, so a settings save during it waits.