Protect query execution and support recoverable 10M-row exports

.github/workflows/ci.yml

@@ -41,6 +41,7 @@ jobs:
           cache: pip
       - run: python -m pip install -e '.[test]'
       - run: python scripts/smoke_container.py
+      - run: python scripts/security_container.py
 
   image-security:
     runs-on: ubuntu-latest

README.md

@@ -79,6 +79,10 @@ the separate `/tmp` tmpfs can therefore remain small and constrained.
 
 ## Verification
 
+See [the coordinated production foundations guide](docs/production-foundations.md)
+for compatibility fixtures, leases and quota behavior, token rotation, mTLS,
+administrative purge, joint testing and rollback.
+
 ```bash
 pytest
 ruff check .

compose.yaml

@@ -1,5 +1,6 @@
 services:
   data-query-worker:
+    init: true
     build:
       context: .
     image: omero-data-query-worker:smoke

deploy/Dockerfile.tests

@@ -0,0 +1,8 @@
+ARG WORKER_IMAGE=omero-data-query-worker:smoke
+FROM ${WORKER_IMAGE}
+USER root
+RUN python -m pip install --no-cache-dir 'pytest>=8.4,<10' 'httpx>=0.28,<1'
+COPY tests /app/tests
+COPY scripts/security_inside.py /app/scripts/security_inside.py
+USER 10001:10001
+CMD ["python", "-m", "pytest", "-q", "-p", "no:cacheprovider", "/app/tests"]

deploy/Dockerfile.tests.dockerignore

@@ -0,0 +1,7 @@
+**
+!deploy
+!deploy/Dockerfile.tests
+!tests
+!tests/**
+!scripts
+!scripts/security_inside.py

deploy/compose.large-export.yaml

@@ -0,0 +1,11 @@
+# Opt in after the target environment passes the large-export gate.
+services:
+  data-query-worker:
+    init: true
+    environment:
+      DQW_MAX_RESULT_ROWS: "10000000"
+      DQW_DUCKDB_MEMORY_LIMIT: "256MB"
+      DQW_MAX_RESULT_BYTES: "2147483648"
+      DQW_QUERY_TIMEOUT_SECONDS: "900"
+      DQW_INGESTION_TIMEOUT_SECONDS: "1800"
+      DQW_RESULT_TTL_SECONDS: "86400"

deploy/compose.mtls.yaml

@@ -0,0 +1,24 @@
+# Use with compose.yaml. Neither service publishes a host port.
+services:
+  data-query-worker:
+    environment:
+      DQW_TOKEN_KEYRING_FILE: /run/query-secrets/tokens.json
+    volumes:
+      - ${DQW_SECRETS_DIR:?Set DQW_SECRETS_DIR}:/run/query-secrets:ro
+  query-tls:
+    image: nginxinc/nginx-unprivileged:1.28-alpine
+    read_only: true
+    cap_drop: [ALL]
+    security_opt: [no-new-privileges:true]
+    pids_limit: 64
+    mem_limit: 128m
+    tmpfs:
+      - /tmp:size=16m,mode=1777
+    volumes:
+      - ./deploy/nginx-mtls.conf:/etc/nginx/conf.d/default.conf:ro
+      - ${DQW_TLS_DIR:?Set DQW_TLS_DIR}:/run/query-tls:ro
+    networks: [worker-private, query-broker]
+    depends_on: [data-query-worker]
+networks:
+  query-broker:
+    internal: true

deploy/nginx-mtls.conf

@@ -0,0 +1,22 @@
+server {
+    listen 8443 ssl;
+    server_name query-tls;
+    ssl_certificate /run/query-tls/server.crt;
+    ssl_certificate_key /run/query-tls/server.key;
+    ssl_client_certificate /run/query-tls/ca.crt;
+    ssl_verify_client on;
+    ssl_protocols TLSv1.2 TLSv1.3;
+    # Do not log result URLs, SQL bodies, Authorization, or client credentials.
+    access_log off;
+    client_max_body_size 21g;
+    client_body_timeout 1800s;
+    location / {
+        proxy_pass http://data-query-worker:8080;
+        proxy_request_buffering off;
+        proxy_buffering off;
+        proxy_read_timeout 1800s;
+        proxy_send_timeout 1800s;
+        proxy_set_header Host $host;
+        proxy_set_header X-Request-ID $http_x_request_id;
+    }
+}

docs/production-foundations.md

@@ -0,0 +1,86 @@
+# Coordinated release: worker 0.2.0 / Analysis 0.14.0
+
+Analysis is the OMERO-aware gateway. The worker remains read-only and has no OMERO credentials or write API. This release retains `omero-data-query-worker-v1`, existing routes, typed `$name` parameters, preview rules and CSV bytes. Analysis retains `omero-data-query-v1`, result tokens, local/remote policy, notebook `ctx.query()` and saved Method bindings v1/v2. Saving is additive and depends on `features.result_provenance_v1`.
+
+## Compatibility contract
+
+The frozen worker source is commit `2115594`; the frozen Analysis broker is from `4db7f8b` on `analysis_integration`. `tests/fixtures/compatibility` contains the contract and original broker. The Analysis HTTP harness runs both brokers against both worker implementations, exercising DuckDB, SQLite, SQLite3, CSV, schema, typed parameters, Unicode/quoting, empty results, downloads, hit/miss/bypass and reuse after rollback. It starts real HTTP processes and separate caches, using the installed dependency set. It is not a substitute for testing every historical engine build.
+
+Tested engines: DuckDB 1.5.5, sqlglot 28.10.1, SQLite 3.45.3 (Windows) and 3.46.1 (Linux container). The retained local worker 0.1.0 image also contains DuckDB 1.5.5/sqlglot 28.10.1/SQLite 3.46.1. SQLite3 is a filename alias for SQLite. The existing browser runtime uses DuckDB 1.5.1. Supported files are standalone databases readable by the selected engine and UTF-8 CSV accepted by the existing ingestion rules; external files, extensions, attached databases and executable database features are outside the supported format. Copying an uncheckpointed database without its journal is unsupported. Original source bytes are never rewritten.
+
+Actual engine, parser, worker and policy versions, effective query limits, source/schema digests and typed parameters participate in result identity. Legacy manifests are readable; missing execution provenance causes query recomputation. Startup discards legacy entries lacking the digest needed for recovery validation, including converted CSV sources without a converted-database digest. Rollback can reuse readable sources and recompute results under the previous key; do not downgrade DuckDB separately without testing its file compatibility.
+
+## Resource and trust boundaries
+
+Use exactly one service process per cache volume. An OS lock rejects a second process. Source leases span execution; result leases span sending the file, including failures/disconnects. Cleanup, quota eviction and administrative purge skip leases. Duplicate deterministic queries and exhausted ingestion/query slots return 429 instead of accumulating lock waiters.
+
+Upload admission happens before multipart parsing. Spools, source copies, CSV conversion output, manifests and result data count toward storage admission. Source and result quotas remain separate. A source may require its spool, immutable copy and converted database simultaneously. If safe eviction cannot make space, the request fails with 507. Result rows are serialized with an exact UTF-8 byte check before each write, including headers and individual oversized values. Subprocess conversion files are observed periodically; use a dedicated filesystem quota in production to cap transient conversion growth and protect unrelated services from a native-engine failure.
+
+Ingestion runs off the async loop and defaults to one simultaneous ingestion; query concurrency remains four. Subprocess limits, cancellation monitoring and `finally` cleanup release pipes, processes, uploads, staging paths and slots. Engine children receive credential-free settings and an allowlisted environment before processing sources or SQL. They still run under the service UID: native-code compromise is contained by the container/host boundaries, not by a separate per-query security identity. Keep secrets and cache inaccessible to other users, worker egress blocked, root filesystem read-only, capabilities dropped, no-new-privileges enabled and cgroup memory/PID limits enforced.
+
+Cache storage contains plaintext scientific data. It is disposable, not an audit store or durable result archive. Persist promoted results and provenance in OMERO. No SQL or parameter values are included in metrics. The Analysis audit records identity, context, hashes, counts, duration and outcome; the full recipe is deliberately stored only in the protected provenance annotation.
+
+Source and result publication now fsyncs the staged files and directories, renames the entry, then fsyncs both parents before returning. On restart, a service/engine lock fences surviving work, incomplete staging is removed immediately and committed file checksums/lengths are verified before readiness. Corrupt entries are discarded for recomputation. Linux execution children install a parent-death signal. Each DuckDB query has a private spill directory inside its quota-accounted staging path, preventing concurrent queries from colliding in a source-adjacent spill directory.
+
+## Optional 10M export profile
+
+The [completed joint release evidence](https://github.com/NL-BioImaging/OMERO.Analysis/blob/297f5c44ed1b0501a6c19444a5eab7f9e8d0f686/docs/testing/query-large-export-2026-09-08.json) validates 540 10M requests: 468 successes, 72 expected admission rejections and zero unexpected failures. All 72 single-client 1M/4M regression requests also pass. Normal runs have no OOM kills and stay within cache quotas. Separate engine-OOM, exact row/byte boundary, live OMERO save and VM power-cut gates pass. See the [tested implementation pair](https://github.com/NL-BioImaging/OMERO.Analysis/blob/297f5c44ed1b0501a6c19444a5eab7f9e8d0f686/docs/testing/query-version-pair.json) for exact commits, engine versions and image/test scope.
+
+`deploy/compose.large-export.yaml` raises the row ceiling to 10,000,000, CSV byte limit to 2 GiB, query timeout to 900 seconds and ingestion timeout to 1,800 seconds. Normal defaults remain unchanged. Apply the matching Analysis profile, including its dedicated CSV promotion limit and web/proxy timeouts. The tested two-CPU, 2 GiB, four-query container uses 256 MB per DuckDB engine; four 512 MB engines exceeded the container memory in testing. Wide results must still fit the byte/time/storage limits or fail explicitly.
+
+The coordinated Analysis [large-export/recovery runbook](https://github.com/NL-BioImaging/OMERO.Analysis/blob/297f5c44ed1b0501a6c19444a5eab7f9e8d0f686/docs/data-query-large-export-recovery.md) provides reproducible 1M/4M/10M fixtures, real HTTP capacity tests, exact 10M/2-GiB boundary checks, real DuckDB/SQLite/CSV OOM, whole-worker OOM, active-query restart and isolated Hyper-V power-cut gates. Real OMERO saved the same verified 609,383,573-byte 10M CSV for all three formats. Analysis journals and an explicit administrator command reconcile interrupted promotions; the worker remains read-only and OMERO-agnostic.
+
+## Operator controls
+
+Additional environment settings:
+
+| Setting | Default / purpose |
+| --- | --- |
+| `DQW_MAX_CONCURRENT_INGESTIONS` | `1`, includes early upload admission |
+| `DQW_CLEANUP_INTERVAL_SECONDS` | `60` |
+| `DQW_TOKEN_KEYRING_FILE` | optional atomically replaced JSON keyring |
+
+`GET /health/live` is independent of storage. Readiness checks writable temporary storage, physical free space, shutdown state and headroom after safe eviction. `GET /v1/metrics` requires bearer authentication and exposes request durations, admission rejection, active operations, cache counters/bytes, failures and subprocess exits. Counters reset on restart; scrape externally. Periodic cleanup removes expired inactive entries and abandoned temporary directories older than twice the larger execution timeout.
+
+Run the local administrative command inside the running worker container:
+
+```sh
+omero-data-query-worker purge --source-id src_<digest>
+omero-data-query-worker purge --scope-id <opaque-scope> --apply
+omero-data-query-worker purge --result-id res_<digest> --apply
+```
+
+The default is dry-run. Exactly one selector is required. Source/scope selection includes associated results. A busy selection is reported without partial deletion. This command contacts the live lease manager over loopback; do not manipulate active cache directories with shell deletion tools.
+
+## Transport and overlapping credentials
+
+Legacy private-network `DQW_API_TOKEN` remains supported. TLS is optional and must not be switched on automatically during a mixed-version upgrade. For rotation, mount a **directory**, set `DQW_TOKEN_KEYRING_FILE=/run/query-secrets/tokens.json`, and atomically rename a new JSON file within that directory. Format: `{"tokens":["old credential at least 16 characters","new credential at least 16 characters"]}`. Never commit real values. Mounting a single file may pin its old inode across host-side atomic replacement. Malformed keyrings fail closed. The environment token is accepted alongside the ring; remove it from deployment configuration when retiring that credential.
+
+`deploy/compose.mtls.yaml` and `deploy/nginx-mtls.conf` supply an optional private mTLS reverse proxy. Combine with `compose.yaml` as a starting profile and replace the smoke token/image with deployment secrets and a pinned candidate image. Mount server.crt, server.key and ca.crt; the server SAN must cover `query-tls`. Attach Analysis only to the broker-facing network. Neither service publishes a host port. Analysis supplies a CA bundle and client certificate/key. Retain bearer authentication behind mTLS. The base compose file is a smoke fixture, not production credential provisioning.
+
+## Reproducible gates and rollout
+
+```sh
+python -m pip install -e '.[test]'
+python -m pytest
+ruff check .
+ruff format --check .
+mypy src
+python scripts/smoke_container.py
+python scripts/security_container.py
+# In the adjacent Analysis checkout (its test dependencies installed):
+DQW_RUN_INTEGRATION=1 python -m pytest tests/test_worker_contract.py
+python scripts/test_local_omero_queries.py
+python scripts/test_query_mtls.py
+python scripts/benchmark_query_capacity.py --rows 100000 --repeats 3
+```
+
+PowerShell: `$env:DQW_RUN_INTEGRATION='1'` before pytest. Set `DQW_REPO`/`DQW_PYTHON` for non-adjacent checkouts. The Linux gate uses disposable containers for regression, read-only filesystem, blocked egress, PID pressure and OOM probes; tests also cover real query timeout/crash/disconnect, corrupt sources, storage exhaustion and leased eviction. The allocator OOM probe validates the cgroup boundary; it does not prove recovery for every possible engine OOM. See the Analysis release document for live permission coverage and remaining acceptance boundaries.
+
+1. Retain both previous images/configuration and back up deployment configuration. Record exact candidate image digests and engine versions. Keep audit/state volumes separate from cache.
+2. Upgrade only the worker. Exercise existing Analysis queries, schema and downloads.
+3. Upgrade Analysis from `analysis_integration`. Check legacy notebooks and Methods, then verify exact CSV promotion and provenance.
+4. Enable the chosen saving/transport deployment configuration after the permission, container and compatibility gates pass on the target environment. Test overlapping credentials before removing the old credential.
+5. To roll back, restore the previous Analysis and/or worker image with its matching transport configuration. Keep OMERO annotations, audit/state and cache volumes. Do not delete caches to force rollback; allow safe reuse/recomputation. Old Analysis ignores additive metadata and offers its existing manual result saving.
+
+Production rollout remains gated on target-environment acceptance, including realistic large exports and restart/fault behavior. These changes do not introduce asynchronous jobs, cancellation APIs, Arrow/Parquet, multi-worker routing, OMERO.Tables materialization or grid discovery.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "omero-data-query-worker"
-version = "0.1.1"
+version = "0.2.0"
 description = "A hardened, OMERO-agnostic service for bounded queries over DuckDB, SQLite, and CSV sources"
 readme = "README.md"
 requires-python = ">=3.11,<3.14"

scripts/security_container.py

@@ -0,0 +1,91 @@
+"""Linux release gate. All destructive fault probes use new disposable containers."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import uuid
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def run(*args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(["docker", *args], cwd=ROOT, check=check, text=True)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--image", default="omero-data-query-worker:smoke")
+    args = parser.parse_args()
+    image = "dqw-security-tests:" + uuid.uuid4().hex[:12]
+    run(
+        "build",
+        "-f",
+        "deploy/Dockerfile.tests",
+        "--build-arg",
+        f"WORKER_IMAGE={args.image}",
+        "-t",
+        image,
+        ".",
+    )
+    security = [
+        "--network",
+        "none",
+        "--read-only",
+        "--cap-drop",
+        "ALL",
+        "--security-opt",
+        "no-new-privileges",
+        "--tmpfs",
+        "/tmp:size=256m,mode=1777",
+    ]
+    try:
+        run("run", "--rm", *security, "--memory", "2g", "--pids-limit", "128", image)
+        for case in ("isolation", "pids"):
+            run(
+                "run",
+                "--rm",
+                *security,
+                "--memory",
+                "256m",
+                "--pids-limit",
+                "32",
+                image,
+                "python",
+                "scripts/security_inside.py",
+                case,
+            )
+        name = "dqw-oom-" + uuid.uuid4().hex[:12]
+        try:
+            result = run(
+                "run",
+                "--name",
+                name,
+                *security,
+                "--memory",
+                "96m",
+                "--memory-swap",
+                "96m",
+                "--pids-limit",
+                "32",
+                image,
+                "python",
+                "scripts/security_inside.py",
+                "oom",
+                check=False,
+            )
+            state = json.loads(subprocess.check_output(["docker", "inspect", name], text=True))[0][
+                "State"
+            ]
+            assert result.returncode == 137 and state["OOMKilled"], state
+        finally:
+            run("rm", "-f", "-v", name, check=False)
+        print("Linux security gates passed: tests, filesystem, egress, PID, OOM")
+    finally:
+        run("image", "rm", image, check=False)
+
+
+if __name__ == "__main__":
+    main()

scripts/security_inside.py

@@ -0,0 +1,59 @@
+"""Bounded OS fault probes, executed only in disposable containers."""
+
+from __future__ import annotations
+
+import argparse
+import errno
+import json
+import os
+import socket
+import subprocess
+import sys
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("case", choices=["isolation", "pids", "oom"])
+    case = parser.parse_args().case
+    if case == "oom":
+        # Cgroup budget is 96 MiB; this container is expected to be OOM killed.
+        bytearray(512 * 1024 * 1024)
+        raise AssertionError("OOM limit was not enforced")
+    if case == "isolation":
+        assert os.getuid() != 0
+        try:
+            with open("/app/forbidden", "w") as handle:
+                handle.write("unexpected")
+        except OSError as exc:
+            assert exc.errno in (errno.EROFS, errno.EACCES)
+        else:
+            raise AssertionError("Root filesystem is writable")
+        try:
+            socket.create_connection(("1.1.1.1", 443), timeout=1)
+        except OSError:
+            pass
+        else:
+            raise AssertionError("Unexpected network egress")
+    if case == "pids":
+        children = []
+        try:
+            for _ in range(64):
+                try:
+                    children.append(
+                        subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
+                    )
+                except OSError as exc:
+                    assert exc.errno == errno.EAGAIN
+                    break
+            else:
+                raise AssertionError("PID limit was not enforced")
+        finally:
+            for child in children:
+                child.terminate()
+            for child in children:
+                child.wait(timeout=5)
+    print(json.dumps({"case": case, "status": "passed"}))
+
+
+if __name__ == "__main__":
+    main()

src/omero_data_query_worker/__init__.py

@@ -1,3 +1,3 @@
 """Hardened tabular data query worker."""
 
-__version__ = "0.1.0"
+__version__ = "0.2.0"