From 685437ec66c140135514ffece8d2d401565f2b3d Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 8 Jun 2026 16:19:47 -0400 Subject: [PATCH 01/24] Fix WES state mapping for completed/materializing invocations completed invocations reported WES UNKNOWN (unmapped); map to COMPLETE. Add requires_materialization -> INITIALIZING. Strengthen test_wes_state_mapping: wait for completion, assert COMPLETE. Co-Authored-By: Claude Opus 4.8 (1M context) --- lib/galaxy/webapps/galaxy/services/wes.py | 2 ++ lib/galaxy_test/api/test_wes.py | 22 +++++++--------------- 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/lib/galaxy/webapps/galaxy/services/wes.py b/lib/galaxy/webapps/galaxy/services/wes.py index 16918f910e9..8ed1a839275 100644 --- a/lib/galaxy/webapps/galaxy/services/wes.py +++ b/lib/galaxy/webapps/galaxy/services/wes.py @@ -71,8 +71,10 @@ log = logging.getLogger(__name__) # Map Galaxy workflow invocation states to WES states GALAXY_TO_WES_STATE = { "new": State.QUEUED, + "requires_materialization": State.INITIALIZING, "ready": State.INITIALIZING, "scheduled": State.RUNNING, + "completed": State.COMPLETE, "failed": State.EXECUTOR_ERROR, "cancelled": State.CANCELED, "cancelling": State.CANCELING, diff --git a/lib/galaxy_test/api/test_wes.py b/lib/galaxy_test/api/test_wes.py index 7ded9b0621e..1b601595547 100644 --- a/lib/galaxy_test/api/test_wes.py +++ b/lib/galaxy_test/api/test_wes.py @@ -386,22 +386,14 @@ steps: history_id=history_id, ) - # Get status via WES API - status = self._get_run_status_validated(invocation_id) + # A successfully finished invocation (Galaxy state "completed") must map to + # the terminal WES state COMPLETE - not UNKNOWN - so clients can poll for done. + self.workflow_populator.wait_for_invocation_and_completion(invocation_id, assert_ok=True) + invocation = self.workflow_populator.get_invocation(invocation_id) + assert invocation["state"] == "completed" - # Verify state is properly mapped - # After waiting for completion, should be COMPLETE or similar - assert status["state"] in [ - "QUEUED", - "INITIALIZING", - "RUNNING", - "PAUSED", - "COMPLETE", - "EXECUTOR_ERROR", - "SYSTEM_ERROR", - "CANCELED", - "CANCELING", - ] + status = self._get_run_status_validated(invocation_id) + assert status["state"] == "COMPLETE" def test_wes_error_handling_missing_workflow_type(self): """Test error handling when workflow_type is missing.""" From 1981c78e8459813447cd2283e9e39a03f5e8c952 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 8 Jun 2026 16:20:14 -0400 Subject: [PATCH 02/24] Add developer GA4GH WES API guide Client walkthrough (curl) of submit/monitor over WES v1.0.0: state mapping, gxworkflow:// scheme, engine params, tasks, pagination, errors, limitations. Cross-link admin <-> dev docs; point at gxy-wes / gxy-wes-bioblend examples. Co-Authored-By: Claude Opus 4.8 (1M context) --- doc/source/admin/ga4gh.md | 22 +- doc/source/dev/ga4gh_wes.md | 444 ++++++++++++++++++++++++++++++++++++ doc/source/dev/index.rst | 1 + 3 files changed, 458 insertions(+), 9 deletions(-) create mode 100644 doc/source/dev/ga4gh_wes.md diff --git a/doc/source/admin/ga4gh.md b/doc/source/admin/ga4gh.md index faed513cc97..fdfca23f38a 100644 --- a/doc/source/admin/ga4gh.md +++ b/doc/source/admin/ga4gh.md @@ -1,8 +1,8 @@ --- myst: - substitutions: - GA4GH_DRS: GA4GH Data Repository Service (DRS) - GA4GH_WES: GA4GH Workflow Execution Service (WES) + substitutions: + GA4GH_DRS: GA4GH Data Repository Service (DRS) + GA4GH_WES: GA4GH Workflow Execution Service (WES) --- # GA4GH API Support @@ -73,6 +73,7 @@ You should see output like: ``` Verify that: + - `organization.name` and `organization.url` match your configured values - `environment` is set appropriately for your deployment - `id` reflects your `ga4gh_service_id` setting (or sensible defaults if not configured) @@ -82,6 +83,8 @@ Verify that: The {{ GA4GH_WES }} enables external systems to submit and monitor Galaxy workflow executions. For detailed API specifications, see the [GA4GH WES specification](https://ga4gh.github.io/workflow-execution-service-schemas/). +For a client/developer walkthrough of submitting and monitoring runs against this API, see +[Developing Against the GA4GH WES API](../dev/ga4gh_wes.md). ### Workflow Types @@ -140,6 +143,7 @@ You should see output like: ``` Verify that: + - `organization.name` and `organization.url` match your configured values - `environment` is set appropriately for your deployment - `id` reflects your `ga4gh_service_id` setting (or sensible defaults if not configured) @@ -150,12 +154,12 @@ All GA4GH configuration is optional and falls back to sensible defaults based on ### Settings -| Setting | Default | Purpose | -|---------|---------|---------| -| `organization_name` | Reversed hostname | Organization name in service responses | -| `organization_url` | Scheme + hostname from request | Organization website URL | -| `ga4gh_service_id` | Reversed hostname | Service ID in reverse domain format (e.g., `org.example`) | -| `ga4gh_service_environment` | (none) | Environment classifier (e.g., "test", "staging", "production") | +| Setting | Default | Purpose | +| --------------------------- | ------------------------------ | -------------------------------------------------------------- | +| `organization_name` | Reversed hostname | Organization name in service responses | +| `organization_url` | Scheme + hostname from request | Organization website URL | +| `ga4gh_service_id` | Reversed hostname | Service ID in reverse domain format (e.g., `org.example`) | +| `ga4gh_service_environment` | (none) | Environment classifier (e.g., "test", "staging", "production") | ### Complete Configuration Example diff --git a/doc/source/dev/ga4gh_wes.md b/doc/source/dev/ga4gh_wes.md new file mode 100644 index 00000000000..5556de4281a --- /dev/null +++ b/doc/source/dev/ga4gh_wes.md @@ -0,0 +1,444 @@ +--- +myst: + substitutions: + GA4GH_WES: GA4GH Workflow Execution Service (WES) +--- + +# Developing Against the GA4GH WES API + +This is a developer-facing companion to the [admin GA4GH docs](../admin/ga4gh.md). The +admin docs cover configuring and verifying the service; this guide walks an API client +through actually submitting and monitoring Galaxy workflow runs over the {{ GA4GH_WES }} +v1.0.0 wire protocol. + +Every request below targets a Galaxy server and uses Galaxy's workflow formats. The curl +examples were exercised against a live development server; the responses shown are real +(ids shortened/anonymized). + +## Example clients + +If you would rather drive the flow below from a runnable tool than from raw curl, two small +reference clients walk exactly this sequence (discover → stage → submit → poll → outputs → +tasks). Both expose a subcommand-per-endpoint CLI plus a `demo` subcommand that runs the +whole example end to end, and both are published to PyPI so they run straight under +[uv](https://docs.astral.sh/uv/) without installing anything: + +```bash +export GXY_WES_API_KEY=... +uvx gxy-wes demo --galaxy-url http://localhost:8080 # requests-only +uvx gxy-wes-bioblend demo --galaxy-url http://localhost:8080 # BioBlend-backed +``` + +- **[gxy-wes](https://github.com/jmchilton/gxy-wes)** — a hand-rolled `requests` wrapper + with no dependency beyond `requests`, so it stays dependency-light and reads as a literal + transcript of the wire protocol. +- **[gxy-wes-bioblend](https://github.com/jmchilton/gxy-wes-bioblend)** — the same CLI + surface, but every call is routed through a [BioBlend](https://bioblend.readthedocs.io/) + `GalaxyInstance`, showing the shape a BioBlend-based application would use. + +```{important} +These two packages are **illustrative examples, not production-supported projects** — they +exist to make this document executable and to show the API two different ways. Do not build +a product on them. For real workflow-execution tooling use +[Planemo](https://planemo.readthedocs.io/) or talk to the Galaxy API through +[BioBlend](https://bioblend.readthedocs.io/) directly. +``` + +## How Galaxy maps onto WES + +WES is workflow-engine agnostic. Galaxy implements it on top of its existing workflow +invocation machinery, so the WES nouns map onto Galaxy nouns: + +| WES concept | Galaxy concept | +| ------------------ | -------------------------------------------------------------------- | +| `run` / `run_id` | `WorkflowInvocation` (the `run_id` is the encoded invocation id) | +| `task` / `task_id` | An invocation step, or a single job within a collection-mapping step | +| run `outputs` | Invocation outputs (HDAs / HDCAs), decorated with DRS URIs | +| `workflow_type` | A Galaxy workflow format (see below) — **not** CWL/Nextflow/WDL | + +`workflow_type` is one of Galaxy's own two workflow formats: + +- **`gx_workflow_format2`** — the gxformat2 ("Format2") YAML format. CWL-like, hand-authorable. +- **`gx_workflow_ga`** — the native `.ga` JSON format produced by Galaxy's workflow editor. + +`workflow_type_version` is currently accepted as a free-form string (e.g. `"1.0.0"`); the +type itself is what matters, and Galaxy will also auto-detect it from the workflow body +(`class: GalaxyWorkflow` ⇒ format2; a top-level `steps`/`workflow` key ⇒ `.ga`). + +## Base URL and authentication + +All endpoints live under `/ga4gh/wes/v1/`. With a local dev server that is: + +``` +http://localhost:8080/ga4gh/wes/v1/ +``` + +Every endpoint except `service-info` requires authentication. Galaxy uses its normal API +key auth — pass `x-api-key: ` (the WES spec's bearer-token scheme is not used). Get a +key for an existing account with HTTP basic auth: + +```bash +curl -s -u "you@example.com:password" \ + http://localhost:8080/api/authenticate/baseauth +# {"api_key": "317ab17d..."} +``` + +Anonymous and inactive accounts are rejected on submission and on every run-scoped read. + +## 1. Discover the service + +`service-info` is public and advertises the supported formats, engine parameters, and +filesystem protocols. + +```bash +curl -s http://localhost:8080/ga4gh/wes/v1/service-info | jq . +``` + +Trimmed to the WES-relevant fields (the real response also carries +`system_state_counts`, `auth_instructions_url`, `tags`, and the standard GA4GH +`organization`/`version`/`environment` keys): + +```json +{ + "id": "localhost.wes", + "name": "Galaxy WES API", + "type": { "group": "org.ga4gh", "artifact": "wes", "version": "1.0.0" }, + "workflow_type_versions": { + "gx_workflow_ga": { "workflow_type_version": ["1.0.0"] }, + "gx_workflow_format2": { "workflow_type_version": ["1.0.0"] } + }, + "supported_wes_versions": ["1.0.0"], + "supported_filesystem_protocols": ["http", "https", "file", "s3", "gs"], + "workflow_engine_versions": { + "galaxy": { "workflow_engine_version": ["1.0.0"] } + }, + "default_workflow_engine_parameters": [ + { "name": "history_name", "type": "string", "default_value": "" }, + { "name": "history_id", "type": "string", "default_value": "" }, + { + "name": "preferred_object_store_id", + "type": "string", + "default_value": "" + }, + { "name": "use_cached_job", "type": "boolean", "default_value": "false" } + ] +} +``` + +## 2. Provide the workflow and its inputs + +A WES `RunRequest` is `multipart/form-data`. The fields Galaxy honors: + +| Field | Required | Notes | +| --------------------------------------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------- | +| `workflow_type` | yes | `gx_workflow_format2` or `gx_workflow_ga`. Must match the auto-detected type or you get a 400. | +| `workflow_type_version` | yes | Free-form string, e.g. `"1.0.0"`. | +| `workflow_url` | one of url/attachment | A URL Galaxy can fetch (`http(s)`, `s3`, `gs`, `file`, `base64://`) **or** a `gxworkflow://` reference (see below). | +| `workflow_attachment` | one of url/attachment | The workflow file uploaded inline. | +| `workflow_params` | no | JSON object of workflow inputs (see below). | +| `workflow_engine_parameters` | no | JSON object of Galaxy-specific run options (see below). | +| `tags` | no | Accepted but currently not persisted onto the invocation. | +| `workflow_engine` / `workflow_engine_version` | no | Accepted; informational. | + +### `workflow_params` — wiring up inputs + +Galaxy workflow inputs are keyed by the workflow's **input label**, and dataset inputs are +references to data that already exists in Galaxy. Galaxy invokes with `inputs_by="name"`, +so the keys are the workflow input names. + +A dataset input is a `{"src": ..., "id": ...}` reference: + +```json +{ "input1": { "src": "hda", "id": "1e8ab44153008be8" } } +``` + +`src` is `hda` for a single dataset or `hdca` for a dataset collection; `id` is the encoded +content id. Simple parameter inputs (integers, text, booleans) are passed as plain JSON +values under their input name. + +WES has **no data-staging endpoint** of its own. You stage inputs through Galaxy's normal +API first (create a history, upload/fetch datasets) and then reference them here. + +### `workflow_engine_parameters` — Galaxy run options + +A JSON object. Recognized keys (also advertised in `service-info`): + +| Key | Effect | +| --------------------------- | -------------------------------------------------------------------------------- | +| `history_id` | Run into this existing history (encoded id). Otherwise a new history is created. | +| `history_name` | Name for the auto-created history (default `"WES Run"`). | +| `preferred_object_store_id` | Object store for the run's outputs. | +| `use_cached_job` | `"true"`/`"false"` string — reuse equivalent prior job results. | + +### Referencing a workflow already in Galaxy: `gxworkflow://` + +Galaxy adds a non-standard URL scheme so you don't have to ship the workflow body on every +submission. Use it as `workflow_url`: + +``` +gxworkflow:// # the StoredWorkflow (latest version) +gxworkflow://?instance=true # a specific Workflow instance +``` + +The caller must own the workflow (or be an admin). With `gxworkflow://`, Galaxy skips +import and invokes the stored workflow directly. `workflow_type` is still required by the +form but is **not** validated against the stored workflow in this case — the +"must match the auto-detected type or 400" check only applies to inline +`workflow_attachment` / fetched `workflow_url` submissions. + +## 3. Submit the run + +Staging step (Galaxy-native, abbreviated) — create a history and upload one dataset: + +```bash +KEY=317ab17d...; BASE=http://localhost:8080 +HIST=$(curl -s -X POST "$BASE/api/histories" -H "x-api-key: $KEY" \ + -H 'Content-Type: application/json' -d '{"name":"WES Tutorial"}' | jq -r .id) +HDA=$(curl -s -X POST "$BASE/api/tools/fetch" -H "x-api-key: $KEY" \ + -F "history_id=$HIST" \ + -F 'targets=[{"destination":{"type":"hdas"},"elements":[{"src":"pasted","paste_content":"line one\nline two\n","ext":"txt","name":"wes_input"}]}]' \ + | jq -r '.outputs[0].id') +``` + +A minimal Format2 workflow (`simple.gxwf.yml`): + +```yaml +class: GalaxyWorkflow +name: Simple Workflow +inputs: + input1: data +outputs: + wf_output_1: + outputSource: first_cat/out_file1 +steps: + first_cat: + tool_id: cat1 + in: + input1: input1 +``` + +Submit it (the WES `POST /runs`): + +```bash +curl -s -X POST "$BASE/ga4gh/wes/v1/runs" -H "x-api-key: $KEY" \ + -F "workflow_type=gx_workflow_format2" \ + -F "workflow_type_version=1.0.0" \ + -F "workflow_params={\"input1\":{\"src\":\"hda\",\"id\":\"$HDA\"}}" \ + -F "workflow_engine_parameters={\"history_id\":\"$HIST\"}" \ + -F "workflow_attachment=@simple.gxwf.yml;type=application/x-yaml" +``` + +```json +{ "run_id": "33b43b4e7093c91f" } +``` + +The `run_id` is the encoded Galaxy invocation id. Batch invocations (multiple invocations +from one request) are rejected by WES. + +## 4. Monitor the run + +Abbreviated status: + +```bash +curl -s "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f/status" -H "x-api-key: $KEY" +# {"run_id":"33b43b4e7093c91f","state":"RUNNING"} +``` + +Full run log — outputs are decorated with DRS URIs so a client can hand them straight to +the DRS API: + +```bash +curl -s "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f" -H "x-api-key: $KEY" | jq . +``` + +```json +{ + "run_id": "33b43b4e7093c91f", + "request": null, + "state": "COMPLETE", + "run_log": null, + "task_logs_url": "/ga4gh/wes/v1/runs/33b43b4e7093c91f/tasks", + "task_logs": null, + "outputs": { + "wf_output_1": { + "src": "hda", + "id": "417e33144b294c21", + "workflow_step_id": 30, + "drs_uri": "drs://drs.localhost:8080/hda-4dcf52aeee371f21" + } + } +} +``` + +Notes on the run log: + +- `request` is always `null` — Galaxy does not persist the original `RunRequest`, so it + cannot be reconstructed. +- `task_logs` is intentionally `null` (deprecated in the WES spec). Use `task_logs_url`. +- Only HDA (single-dataset) outputs get a `drs_uri`; collection (HDCA) outputs carry an + encoded `id` but no DRS URI. + +### State mapping + +WES has a single flat `State` enum (`QUEUED`, `INITIALIZING`, `RUNNING`, `COMPLETE`, +`EXECUTOR_ERROR`, `CANCELING`, `CANCELED`, ...). A Galaxy run is a `WorkflowInvocation`, +whose state is mapped onto that enum (`GALAXY_TO_WES_STATE` in `services/wes.py`): + +| Galaxy invocation state | WES state | Meaning | +| -------------------------- | ---------------- | ------------------------------------------------------- | +| `new` | `QUEUED` | Invocation created, not yet scheduling. | +| `requires_materialization` | `INITIALIZING` | Deferred inputs are being materialized. | +| `ready` | `INITIALIZING` | Ready for a scheduling iteration. | +| `scheduled` | `RUNNING` | Fully scheduled; jobs are running/queued. | +| `completed` | `COMPLETE` | Scheduling done and every job reached a terminal state. | +| `failed` | `EXECUTOR_ERROR` | The invocation itself failed to schedule. | +| `cancelled` | `CANCELED` | Invocation was cancelled. | +| `cancelling` | `CANCELING` | Cancellation requested, in progress. | +| (anything unmapped) | `UNKNOWN` | Fallback; should not occur for current states. | + +Poll `status` (or the run log `state`) until the state is terminal: `COMPLETE` (success), +`EXECUTOR_ERROR` (failure), or `CANCELED` (cancelled). Keep polling while it is any of the +non-terminal states (`QUEUED`, `INITIALIZING`, `RUNNING`, `CANCELING`). + +#### Why `completed` ⇒ `COMPLETE`, even when a job failed + +This is intentional and important: **a failed job does not make the run unsuccessful.** +WES `state` describes the run (the invocation), not individual jobs. Galaxy's `completed` +means scheduling finished and all jobs reached a terminal state (`ok`, `error`, `skipped`, +`paused`, ...) — and a job ending in `error` is frequently part of a perfectly normal, +valid execution path. For example, a workflow can route a step through a _filter-failed_ +collection operation precisely so that some jobs are allowed to fail and the failures are +filtered out downstream; the invocation still completes successfully. + +So: + +- The only invocation state that maps to `EXECUTOR_ERROR` is Galaxy's own `failed` — i.e. + the invocation could not be scheduled — **not** "some job errored." +- Clients should **not** infer run failure from a non-zero per-task `exit_code`. Per-task + `exit_code`s (from the tasks endpoint) are for inspecting individual steps, not for + deciding whether the run as a whole succeeded. +- Determine run success from the run `state` (`COMPLETE`); inspect outputs and per-task + detail for finer-grained results. + +## 5. Tasks (per-step / per-job detail) + +`task_logs_url` points at a paginated task list. A `task_id` is the step `order_index` +(e.g. `"0"`, `"1"`), or `order_index.job_index` (e.g. `"1.0"`, `"1.2"`) for the individual +jobs of a collection-mapping step. + +```bash +curl -s "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f/tasks" -H "x-api-key: $KEY" | jq . +``` + +```json +{ + "task_logs": [ + { + "id": "0", + "name": "input1", + "exit_code": null, + "stdout": null, + "stderr": null + }, + { + "id": "1", + "name": "first_cat", + "exit_code": 0, + "stdout": "/api/jobs/417e33144b294c21/stdout", + "stderr": "/api/jobs/417e33144b294c21/stderr" + } + ], + "next_page_token": null +} +``` + +`stdout` / `stderr` are URLs to plain-text job-output endpoints +(`GET /api/jobs/{job_id}/stdout` and `/stderr`). Steps with no underlying job — input +steps, for example — have `null` `exit_code`, `stdout`, and `stderr` (as `id: "0"` shows +above). Fetch a single task with `GET /ga4gh/wes/v1/runs/{run_id}/tasks/{task_id}`. + +### Pagination + +Both `GET /runs` and `GET /runs/{id}/tasks` use opaque keyset (cursor) tokens, not +offsets. Pass `page_size` (1–100, default 10); if more results exist the response includes +`next_page_token`, which you echo back as `page_token` on the next request. Do not try to +decode or construct tokens yourself. + +```bash +curl -s "$BASE/ga4gh/wes/v1/runs?page_size=2" -H "x-api-key: $KEY" +curl -s "$BASE/ga4gh/wes/v1/runs?page_size=2&page_token=" -H "x-api-key: $KEY" +``` + +`GET /runs` only ever lists the authenticated user's own runs. + +## 6. Cancel a run + +```bash +curl -s -X POST "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f/cancel" -H "x-api-key: $KEY" +# {"run_id": "33b43b4e7093c91f"} +``` + +This requests cancellation of the underlying invocation; the run then transitions through +`CANCELING` to `CANCELED`. + +## Errors + +Errors come back as Galaxy's standard JSON error body (`{"err_msg": ..., "err_code": ...}`) +with a matching HTTP status. The ones a client is most likely to hit: + +| Situation | Status | +| -------------------------------------------------------------------------------- | ------ | +| Missing/invalid required form field (e.g. no `workflow_type`) | 400 | +| `workflow_type` disagrees with the auto-detected type (attachment / fetched URL) | 400 | +| Malformed `gxworkflow://` URI | 400 | +| Submit while anonymous | 403 | +| Submit while logged in but inactive (unactivated account) | 403 | +| Run / task id does not exist | 404 | +| Invalid `task_id` format | 404 | + +Access to resources you do not own is **not** reported uniformly — watch for this: + +- Reading a **run** you don't own → `403` (`AuthenticationRequired`). +- Submitting against a **history** you don't own → `404` (`ObjectNotFound`). +- A `gxworkflow://` reference to a **workflow** you don't own → `403` + (`ItemAccessibilityException`). + +So a `404` on submission can mean "your history id is wrong" _or_ "that history belongs to +someone else"; don't assume `404` always means the object is absent. + +## Endpoint summary + +| Method | Path | Auth | Purpose | +| ------ | --------------------------------------------- | -------- | --------------------------------------- | +| GET | `/ga4gh/wes/v1/service-info` | public | Capabilities and supported formats | +| POST | `/ga4gh/wes/v1/runs` | required | Submit a run (`multipart/form-data`) | +| GET | `/ga4gh/wes/v1/runs` | required | List the user's runs (keyset paginated) | +| GET | `/ga4gh/wes/v1/runs/{run_id}` | required | Full run log + DRS-decorated outputs | +| GET | `/ga4gh/wes/v1/runs/{run_id}/status` | required | Abbreviated `{run_id, state}` | +| POST | `/ga4gh/wes/v1/runs/{run_id}/cancel` | required | Request cancellation | +| GET | `/ga4gh/wes/v1/runs/{run_id}/tasks` | required | Paginated task list | +| GET | `/ga4gh/wes/v1/runs/{run_id}/tasks/{task_id}` | required | Single task detail | + +## Known limitations and gotchas + +- **Run `COMPLETE` reflects the invocation, not individual jobs** — by design, a failed + job does not make the run fail (a failed job can be a normal, valid path, e.g. + filter-failed). Only an invocation that fails to schedule maps to `EXECUTOR_ERROR`. Do + not infer run failure from a per-task `exit_code`. See §4. +- **Inputs must be pre-staged** — there is no WES upload endpoint; stage datasets via the + Galaxy API and reference them by `{src, id}` in `workflow_params`. +- **`request` is never returned** in the run log — the original `RunRequest` is not stored. +- **DRS URIs only on HDA outputs**, not on collection outputs. +- **`tags` are not persisted** to the invocation, and run summaries return `tags: {}`. +- **`service-info` `auth_instructions_url` is a placeholder** (`"TODO"`). +- The generated Pydantic models under `lib/galaxy/schema/wes/` are produced by + `gen.sh` from the upstream GA4GH OpenAPI spec with no Galaxy post-processing — treat + them as regenerable and avoid hand edits. + +## Source pointers + +- Router: `lib/galaxy/webapps/galaxy/api/wes.py` +- Service: `lib/galaxy/webapps/galaxy/services/wes.py` +- Generated models: `lib/galaxy/schema/wes/` (regenerate via `gen.sh`) +- Shared GA4GH service-info builder: `lib/galaxy/webapps/galaxy/services/ga4gh.py` +- API tests (worked examples for every endpoint): `lib/galaxy_test/api/test_wes.py` diff --git a/doc/source/dev/index.rst b/doc/source/dev/index.rst index 23e212d0a63..e248bc5bc61 100644 --- a/doc/source/dev/index.rst +++ b/doc/source/dev/index.rst @@ -20,6 +20,7 @@ A multi-hour long video playlist covering these slides can be found at data_managers data_source data_types + ga4gh_wes faq writing_tests debugging_tests From 9a7f0978021b26f9440f83f98f02c8cc7039ebaf Mon Sep 17 00:00:00 2001 From: JunhaoQiu <56094690+qchiujunhao@users.noreply.github.com> Date: Wed, 17 Jun 2026 11:31:25 -0400 Subject: [PATCH 03/24] Hide generated headers for tabular previews --- .../Tabular/TabularChunkedView.test.ts | 42 +++++++++++++++++++ .../Dataset/Tabular/TabularChunkedView.vue | 1 + lib/galaxy/datatypes/tabular.py | 6 +++ test/unit/data/datatypes/test_tabular.py | 13 ++++++ 4 files changed, 62 insertions(+) create mode 100644 client/src/components/Dataset/Tabular/TabularChunkedView.test.ts diff --git a/client/src/components/Dataset/Tabular/TabularChunkedView.test.ts b/client/src/components/Dataset/Tabular/TabularChunkedView.test.ts new file mode 100644 index 00000000000..24c0fa9b468 --- /dev/null +++ b/client/src/components/Dataset/Tabular/TabularChunkedView.test.ts @@ -0,0 +1,42 @@ +import { getLocalVue } from "@tests/vitest/helpers"; +import { shallowMount } from "@vue/test-utils"; +import axios from "axios"; +import { describe, expect, it, vi } from "vitest"; + +import TabularChunkedView from "./TabularChunkedView.vue"; +import GTable from "@/components/Common/GTable.vue"; + +vi.mock("axios"); +vi.mock("@/onload/loadConfig", () => ({ + getAppRoot: () => "/", +})); + +const localVue = getLocalVue(); + +function mountChunkedView(fileExt: string) { + vi.mocked(axios.get).mockResolvedValue({ data: { ck_data: "", offset: 0, data_line_offset: 0 } }); + return shallowMount(TabularChunkedView as object, { + localVue, + propsData: { + options: { + id: "dataset-id", + file_ext: fileExt, + metadata_columns: 2, + }, + }, + }); +} + +describe("TabularChunkedView", () => { + it("hides the table header for generic tabular datasets", () => { + const wrapper = mountChunkedView("tabular"); + + expect(wrapper.findComponent(GTable).props("hideHeader")).toBe(true); + }); + + it("keeps the table header for CSV datasets", () => { + const wrapper = mountChunkedView("csv"); + + expect(wrapper.findComponent(GTable).props("hideHeader")).toBe(false); + }); +}); diff --git a/client/src/components/Dataset/Tabular/TabularChunkedView.vue b/client/src/components/Dataset/Tabular/TabularChunkedView.vue index 90e13898d37..61830994c39 100644 --- a/client/src/components/Dataset/Tabular/TabularChunkedView.vue +++ b/client/src/components/Dataset/Tabular/TabularChunkedView.vue @@ -200,6 +200,7 @@ onMounted(() => { striped head-variant="dark" :fields="fields" + :hide-header="props.options.file_ext === 'tabular'" :items="tableRows" :load-more-loading="loading" /> diff --git a/lib/galaxy/datatypes/tabular.py b/lib/galaxy/datatypes/tabular.py index 1e4f0e0cec0..cc6e13c221f 100644 --- a/lib/galaxy/datatypes/tabular.py +++ b/lib/galaxy/datatypes/tabular.py @@ -390,6 +390,12 @@ class Tabular(TabularData): def get_column_names(self, first_line: str) -> Optional[list[str]]: return None + def make_html_peek_header(self, dataset: DatasetProtocol, *args, **kwargs) -> str: + # Only generic tabular loses generated headers; subclasses may define real column labels. + if type(self) is Tabular: + return "" + return super().make_html_peek_header(dataset, *args, **kwargs) + def set_meta( self, dataset: DatasetProtocol, diff --git a/test/unit/data/datatypes/test_tabular.py b/test/unit/data/datatypes/test_tabular.py index c98fc1c6f7e..fb6794c32ed 100644 --- a/test/unit/data/datatypes/test_tabular.py +++ b/test/unit/data/datatypes/test_tabular.py @@ -112,3 +112,16 @@ def test_tabular_column_types_override(): assert dataset.metadata.columns == 6 assert dataset.metadata.delimiter == "\t" assert not hasattr(dataset.metadata, "column_names") + + +def test_tabular_display_peek_does_not_render_table_header(): + dataset = MockDataset(id=1) + dataset.peek = "question_id\tcurator_name\nensembl-grab-q1\tLG\n" + dataset.metadata.columns = 2 + dataset.metadata.delimiter = "\t" + + html = Tabular().display_peek(dataset) # type: ignore [arg-type] + + assert "" not in html + assert "question_id" in html + assert "curator_name" in html From 1fafefea30e54f5d2d20d8d98c3fab914fe70c4b Mon Sep 17 00:00:00 2001 From: JunhaoQiu <56094690+qchiujunhao@users.noreply.github.com> Date: Wed, 17 Jun 2026 12:16:53 -0400 Subject: [PATCH 04/24] Fix tabular preview test typing --- test/unit/data/datatypes/test_tabular.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/unit/data/datatypes/test_tabular.py b/test/unit/data/datatypes/test_tabular.py index fb6794c32ed..4a3d1a7faf5 100644 --- a/test/unit/data/datatypes/test_tabular.py +++ b/test/unit/data/datatypes/test_tabular.py @@ -116,9 +116,9 @@ def test_tabular_column_types_override(): def test_tabular_display_peek_does_not_render_table_header(): dataset = MockDataset(id=1) - dataset.peek = "question_id\tcurator_name\nensembl-grab-q1\tLG\n" - dataset.metadata.columns = 2 - dataset.metadata.delimiter = "\t" + setattr(dataset, "peek", "question_id\tcurator_name\nensembl-grab-q1\tLG\n") + setattr(dataset.metadata, "columns", 2) + setattr(dataset.metadata, "delimiter", "\t") html = Tabular().display_peek(dataset) # type: ignore [arg-type] From 5920b76fcbaac26b3f4b0859a9482c0b8239ae86 Mon Sep 17 00:00:00 2001 From: JunhaoQiu <56094690+qchiujunhao@users.noreply.github.com> Date: Wed, 17 Jun 2026 12:35:24 -0400 Subject: [PATCH 05/24] Fix tabular preview test lint --- test/unit/data/datatypes/test_tabular.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/test/unit/data/datatypes/test_tabular.py b/test/unit/data/datatypes/test_tabular.py index 4a3d1a7faf5..b0c3288d61a 100644 --- a/test/unit/data/datatypes/test_tabular.py +++ b/test/unit/data/datatypes/test_tabular.py @@ -1,4 +1,8 @@ import tempfile +from typing import ( + Any, + cast, +) from galaxy.datatypes.tabular import ( MAX_DATA_LINES, @@ -116,9 +120,11 @@ def test_tabular_column_types_override(): def test_tabular_display_peek_does_not_render_table_header(): dataset = MockDataset(id=1) - setattr(dataset, "peek", "question_id\tcurator_name\nensembl-grab-q1\tLG\n") - setattr(dataset.metadata, "columns", 2) - setattr(dataset.metadata, "delimiter", "\t") + dataset_for_peek = cast(Any, dataset) + metadata = cast(Any, dataset.metadata) + dataset_for_peek.peek = "question_id\tcurator_name\nensembl-grab-q1\tLG\n" + metadata.columns = 2 + metadata.delimiter = "\t" html = Tabular().display_peek(dataset) # type: ignore [arg-type] From 9947c087eb58f53012549ce87b50acaf23ac4ee1 Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Tue, 14 Jul 2026 13:27:15 +0200 Subject: [PATCH 06/24] implement multipart upload for invenio data source --- .../sample/file_sources_conf.yml.sample | 12 +- lib/galaxy/files/sources/_rdm.py | 6 + lib/galaxy/files/sources/invenio.py | 222 +++++++++++++++++- test/unit/files/test_invenio_multipart.py | 118 ++++++++++ 4 files changed, 352 insertions(+), 6 deletions(-) create mode 100644 test/unit/files/test_invenio_multipart.py diff --git a/lib/galaxy/config/sample/file_sources_conf.yml.sample b/lib/galaxy/config/sample/file_sources_conf.yml.sample index 5b46c69cb07..560c2060519 100644 --- a/lib/galaxy/config/sample/file_sources_conf.yml.sample +++ b/lib/galaxy/config/sample/file_sources_conf.yml.sample @@ -15,7 +15,7 @@ # for accessing passwords stored in a vault: # password: ${user.user_vault.read_secret('preferences/owncloud/password')} - # By default, the plugin will use temp files to avoid loading entire files into memory. + # By default, the plugin will use temp files to avoid loading entire files into memory. # You can change the directory here or omit to use the default temp directory. temp_path: /your/temp/path # Set writable to true if you have write access to this source @@ -84,8 +84,8 @@ enable_resume: true # Enable resume for interrupted transfers # SSH private key for authentication (download ENA key from: https://www.ebi.ac.uk/ena/browser/about/ascp) # Embed the key content directly in the configuration (required - not file paths) - # Note: SSH key content is required because Galaxy jobs often run on clusters that don't - # mount Galaxy's root or configuration directories. The configuration block is copied + # Note: SSH key content is required because Galaxy jobs often run on clusters that don't + # mount Galaxy's root or configuration directories. The configuration block is copied # to the job's directory, but referenced key paths wouldn't be accessible. ssh_key_content: | -----BEGIN RSA PRIVATE KEY----- @@ -259,6 +259,12 @@ # token: ${user.preferences['invenio_sandbox|token']} # Alternatively use this for retrieving the token from user preferences instead of the Vault public_name: ${user.preferences['invenio_sandbox|public_name']} writable: true + # Enable multipart upload for files of size >= threshold (values in MB, optional, disabled by default) + # multipart_threshold: 100 + # Part size for multipart uploads (values in MB, optional, defaults to 5 MB minimum) + # multipart_chunk_size: 50 + # Default resource type for new records (optional, defaults to "dataset") + # default_resource_type: - type: zenodo id: zenodo diff --git a/lib/galaxy/files/sources/_rdm.py b/lib/galaxy/files/sources/_rdm.py index 2a1e3ae3542..e3a142286ac 100644 --- a/lib/galaxy/files/sources/_rdm.py +++ b/lib/galaxy/files/sources/_rdm.py @@ -25,11 +25,17 @@ log = logging.getLogger(__name__) class RDMFileSourceTemplateConfiguration(BaseFileSourceTemplateConfiguration): token: Optional[Union[str, TemplateExpansion]] = None public_name: Optional[Union[str, TemplateExpansion]] = None + multipart_threshold: Optional[Union[int, TemplateExpansion]] = None # MB + multipart_chunk_size: Optional[Union[int, TemplateExpansion]] = None # MB + default_resource_type: Optional[str] = None class RDMFileSourceConfiguration(BaseFileSourceConfiguration): token: Optional[str] = None public_name: Optional[str] = None + multipart_threshold: Optional[int] = None # MB + multipart_chunk_size: Optional[int] = None # MB + default_resource_type: Optional[str] = None class ContainerAndFileIdentifier(NamedTuple): diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 8d24eba4c5b..1fa9413340f 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -1,6 +1,13 @@ import datetime import json +import logging +import math +import os import re +from concurrent.futures import ( + as_completed, + ThreadPoolExecutor, +) from typing import ( Any, cast, @@ -9,6 +16,8 @@ from typing import ( ) from urllib.parse import quote +log = logging.getLogger(__name__) + from typing_extensions import ( TypedDict, ) @@ -102,6 +111,53 @@ class RecordLinks(TypedDict): reserve_doi: str +# AWS S3 multipart default limits (used by Invenio RDM) +MIN_UPLOAD_PART_SIZE = 5 * 1024 * 1024 # 5 MiB +MAX_UPLOAD_PART_SIZE = 5 * 1024**3 # 5 GiB +MAX_UPLOAD_PARTS = 10_000 + + +def calculate_multipart_params(file_size: int, preferred_part_size: int | None = None) -> tuple[int, int]: + """Calculate optimal parts count and part size for multipart upload. + + Args: + file_size: Total file size in bytes + preferred_part_size: Preferred part size in bytes (optional) + + Returns: + Tuple of (parts_count, part_size) + + Note: + Maximum uploadable file size is MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB). + Files larger than this will still return valid params but would fail server-side. + """ + if file_size == 0: + return 1, 0 + + # Start with preferred or minimum part size + part_size = preferred_part_size or MIN_UPLOAD_PART_SIZE + + # Ensure part_size is within bounds + part_size = max(part_size, MIN_UPLOAD_PART_SIZE) + part_size = min(part_size, MAX_UPLOAD_PART_SIZE) + + # Calculate parts needed + parts = math.ceil(file_size / part_size) + + # If too many parts, increase part size (up to max) + while parts > MAX_UPLOAD_PARTS and part_size < MAX_UPLOAD_PART_SIZE: + part_size = min(part_size * 2, MAX_UPLOAD_PART_SIZE) + parts = math.ceil(file_size / part_size) + + # For extremely large files, cap parts at MAX_UPLOAD_PARTS + # This means part_size may effectively be larger than calculated + # but such files would likely fail server-side anyway + if parts > MAX_UPLOAD_PARTS: + parts = MAX_UPLOAD_PARTS + + return parts, part_size + + class InvenioRecord(TypedDict): id: str title: str @@ -305,12 +361,13 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): ) -> dict[str, Any]: today = datetime.date.today().isoformat() creator = self._get_creator_from_public_name(public_name) + resource_type_id = context.config.default_resource_type or "dataset" create_record_request = { "files": {"enabled": True}, "metadata": { "title": title, "publication_date": today, - "resource_type": {"id": "dataset"}, + "resource_type": {"id": resource_type_id}, "creators": [ creator, ], @@ -331,6 +388,29 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): file_path: str, context: FilesSourceRuntimeContext[RDMFileSourceConfiguration], ): + file_size = os.path.getsize(file_path) + + threshold_mb = context.config.multipart_threshold + # Convert threshold from MB to bytes (config value is always in MB) + threshold_bytes = threshold_mb * 1024 * 1024 if threshold_mb else None + use_multipart = file_size >= threshold_bytes if threshold_bytes else False + if use_multipart: + self._upload_file_multipart(record_id, filename, file_path, file_size, context) + else: + self._upload_file_single(record_id, filename, file_path, context, file_size) + + def _upload_file_single( + self, + record_id: str, + filename: str, + file_path: str, + context: FilesSourceRuntimeContext[RDMFileSourceConfiguration], + file_size: Optional[int] = None, + ): + """Upload a file using single PUT request.""" + if file_size is None: + file_size = os.path.getsize(file_path) + record = self._get_draft_record(record_id, context) upload_file_url = record["links"]["files"] headers = self._get_request_headers(context, auth_required=True) @@ -346,12 +426,143 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): commit_file_upload_url = file_entry["links"]["commit"] with open(file_path, "rb") as file: response = requests.put(upload_file_content_url, data=file, headers=headers) + # Handle 413 (Payload Too Large) - suggest using multipart upload + if response.status_code == 413: + raise Exception( + f"Failed to upload file '{filename}' ({file_size} bytes): HTTP 413 Payload Too Large. " + f"The server rejected the upload because the file is too large for a single request. " + f"Please configure 'multipart_threshold' in the file source configuration to enable multipart upload for files of this size." + ) self._ensure_response_has_expected_status_code(response, 200) # Commit file upload response = requests.post(commit_file_upload_url, headers=headers) self._ensure_response_has_expected_status_code(response, 200) + def _upload_file_multipart( + self, + record_id: str, + filename: str, + file_path: str, + file_size: int, + context: FilesSourceRuntimeContext[RDMFileSourceConfiguration], + ): + """Upload a file using multipart upload. + + Flow: + 1. Calculate parts/part_size + 2. POST with transfer metadata + 3. Server returns links.parts[] with URL for each part + 4. Upload parts (parallel for > 2 parts) + 5. POST to commit URL + """ + preferred_part_size_mb = context.config.multipart_chunk_size + # Convert chunk size from MB to bytes (config value is always in MB) + preferred_part_size = preferred_part_size_mb * 1024 * 1024 if preferred_part_size_mb else None + num_parts, part_size = calculate_multipart_params(file_size, preferred_part_size) + + log.info(f"Multipart upload: {num_parts} parts of {part_size} bytes each for '{filename}'") + + record = self._get_draft_record(record_id, context) + upload_file_url = record["links"]["files"] + headers = self._get_request_headers(context, auth_required=True) + + file_metadata = { + "key": filename, + "size": file_size, + "transfer": { + "type": "M", + "parts": num_parts, + "part_size": part_size, + }, + } + response = requests.post(upload_file_url, json=[file_metadata], headers=headers) + self._ensure_response_has_expected_status_code(response, 201) + + # Get part upload URLs from response + entries = response.json()["entries"] + file_entry = next(entry for entry in entries if entry["key"] == filename) + commit_url = file_entry["links"]["commit"] + part_links = file_entry.get("links", {}).get("parts", []) + + if len(part_links) != num_parts: + raise Exception( + f"Server returned {len(part_links)} part URLs but expected {num_parts} for file '{filename}'" + ) + + # Sort part links by part number to ensure correct ordering + part_links = sorted(part_links, key=lambda p: p.get("part", 0)) + self._upload_parts(file_path, file_size, part_size, part_links, headers) + response = requests.post(commit_url, json={}, headers=headers) + self._ensure_response_has_expected_status_code(response, 200) + log.info(f"Multipart upload completed for '{filename}'") + + def _upload_parts( + self, + file_path: str, + file_size: int, + part_size: int, + part_links: list[dict], + headers: dict, + ): + """Upload all parts, sequentially for <=2 parts, parallel otherwise.""" + num_parts = len(part_links) + + if num_parts <= 2: + for part_index, part_info in enumerate(part_links): + self._upload_single_part(file_path, file_size, part_size, part_index, part_info) + else: + max_workers = min(4, num_parts) + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = {} + for part_index, part_info in enumerate(part_links): + future = executor.submit( + self._upload_single_part, + file_path, + file_size, + part_size, + part_index, + part_info, + ) + futures[future] = part_index + + for future in as_completed(futures): + part_index = futures[future] + try: + future.result() + except Exception as e: + log.error(f"Failed to upload part {part_index}: {e}") + raise + + def _upload_single_part( + self, + file_path: str, + file_size: int, + part_size: int, + part_index: int, + part_info: dict, + ): + """Upload a single part of a multipart upload.""" + part_url = part_info.get("url") + if not part_url: + raise Exception(f"No URL provided for part {part_index}") + + start_byte = part_index * part_size + end_byte = min(start_byte + part_size, file_size) + part_content_length = end_byte - start_byte + + log.debug(f"Uploading part {part_index}: bytes {start_byte}-{end_byte-1} ({part_content_length} bytes)") + + # Read the entire part into memory and upload + with open(file_path, "rb") as f: + f.seek(start_byte) + part_data = f.read(part_content_length) + + # Use empty headers - presigned URLs are authenticated via query parameters + # Adding Authorization or other headers would invalidate the signature + response = requests.put(part_url, data=part_data) + self._ensure_response_has_expected_status_code(response, 200) + def download_file_from_container( self, container_id: str, @@ -558,11 +769,16 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): def _raise_auth_required(self): raise AuthenticationRequired( - f"Please provide a personal access token in your user's preferences for '{self.plugin.label}'" + f"Access denied. Please make sure you have provided a personal access token in your user's preferences for '{self.plugin.label}'" ) def _get_response_error_message(self, response): - response_json = response.json() + try: + response_json = response.json() + except Exception: + # Response is not JSON, return raw text or status info + return response.text or f"HTTP {response.status_code} error" + error_message = response_json.get("message") if response.status_code == 400 else response.text errors = response_json.get("errors", []) for error in errors: diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py new file mode 100644 index 00000000000..c37dee2eb7b --- /dev/null +++ b/test/unit/files/test_invenio_multipart.py @@ -0,0 +1,118 @@ +"""Unit tests for Invenio multipart upload functionality.""" + +import pytest + +from galaxy.files.sources.invenio import ( + calculate_multipart_params, + MAX_UPLOAD_PART_SIZE, + MAX_UPLOAD_PARTS, + MIN_UPLOAD_PART_SIZE, +) + + +class TestCalculateMultipartParams: + """Tests for calculate_multipart_params function.""" + + def test_calculate_multipart_params_zero_byte(self): + """Zero-byte files should return (1, 0).""" + parts, part_size = calculate_multipart_params(0) + assert parts == 1 + assert part_size == 0 + + def test_calculate_multipart_params_small_file(self): + """Files under 5 MiB should use minimum part size.""" + # 2 MiB file + file_size = 2 * 1024 * 1024 + parts, part_size = calculate_multipart_params(file_size) + assert parts == 1 + assert part_size == MIN_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_medium_file(self): + """Files between 5 MiB and 10 MiB.""" + # 7.5 MiB file + file_size = 7 * 1024 * 1024 + 512 * 1024 + parts, part_size = calculate_multipart_params(file_size) + assert parts == 2 + assert part_size == MIN_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_large_file(self): + """Large files requiring multiple parts.""" + # 25 MiB file + file_size = 25 * 1024 * 1024 + parts, part_size = calculate_multipart_params(file_size) + assert parts == 5 + assert part_size == MIN_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_respects_max_parts(self): + """Very large files should not exceed MAX_UPLOAD_PARTS.""" + # File larger than MAX_UPLOAD_PARTS * MIN_UPLOAD_PART_SIZE + file_size = MAX_UPLOAD_PARTS * MIN_UPLOAD_PART_SIZE + MIN_UPLOAD_PART_SIZE + parts, part_size = calculate_multipart_params(file_size) + assert parts <= MAX_UPLOAD_PARTS + assert part_size >= MIN_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_extremely_large_file(self): + """Extremely large files should hit both MAX limits. + + Note: Files larger than MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB) + cannot be uploaded via multipart, but we cap params rather than fail here. + The upload would fail server-side anyway. + """ + # 100 TiB file - exceeds theoretical maximum (~48.8 TiB) + file_size = 100 * 1024**4 # 100 TiB + parts, part_size = calculate_multipart_params(file_size) + # Parts should be capped at MAX_UPLOAD_PARTS + assert parts == MAX_UPLOAD_PARTS + # Part size should hit max + assert part_size == MAX_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_respects_preferred_part_size(self): + """Should use preferred part size when provided and valid.""" + # 150 MiB file with 100 MiB preferred part size + file_size = 150 * 1024 * 1024 + preferred_part_size = 100 * 1024 * 1024 # 100 MiB + parts, part_size = calculate_multipart_params(file_size, preferred_part_size) + assert parts == 2 + assert part_size == preferred_part_size + + def test_calculate_multipart_params_preferred_too_small(self): + """Should use minimum part size if preferred is too small.""" + # 100 MiB file with 1 MiB preferred part size (too small) + file_size = 100 * 1024 * 1024 + preferred_part_size = 1 * 1024 * 1024 # 1 MiB - too small + parts, part_size = calculate_multipart_params(file_size, preferred_part_size) + assert part_size == MIN_UPLOAD_PART_SIZE # Should be bumped to minimum + + def test_calculate_multipart_params_preferred_exceeds_max(self): + """Should cap at MAX_UPLOAD_PART_SIZE if preferred exceeds it.""" + # Small file with huge preferred part size + file_size = 100 * 1024 * 1024 + preferred_part_size = MAX_UPLOAD_PART_SIZE * 2 # Exceeds max + parts, part_size = calculate_multipart_params(file_size, preferred_part_size) + assert parts == 1 + assert part_size == MAX_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_exact_multiple(self): + """File size that's an exact multiple of part size.""" + # Exactly 3 * MIN_UPLOAD_PART_SIZE + file_size = 3 * MIN_UPLOAD_PART_SIZE + parts, part_size = calculate_multipart_params(file_size) + assert parts == 3 + assert part_size == MIN_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_one_byte_over(self): + """File size one byte over an exact multiple.""" + # 3 * MIN_UPLOAD_PART_SIZE + 1 byte + file_size = 3 * MIN_UPLOAD_PART_SIZE + 1 + parts, part_size = calculate_multipart_params(file_size) + assert parts == 4 # Need 4 parts for 3 full + 1 byte + assert part_size == MIN_UPLOAD_PART_SIZE + + def test_calculate_multipart_params_at_boundary(self): + """Test file at MAX_UPLOAD_PARTS boundary.""" + # Exactly at the boundary where we need to increase part size + file_size = (MAX_UPLOAD_PARTS + 1) * MIN_UPLOAD_PART_SIZE + parts, part_size = calculate_multipart_params(file_size) + assert parts <= MAX_UPLOAD_PARTS + # Part size should have increased + assert part_size > MIN_UPLOAD_PART_SIZE From 64876e39cb9866b8148894a476e003e1509b3a94 Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Tue, 14 Jul 2026 13:42:55 +0200 Subject: [PATCH 07/24] limit the memory usage for large multipart upload --- lib/galaxy/files/sources/invenio.py | 38 +++++++++++++++++++++++------ 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 1fa9413340f..f99dd5b8f20 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -158,6 +158,31 @@ def calculate_multipart_params(file_size: int, preferred_part_size: int | None = return parts, part_size +class _LimitedFileReader: + """File-like wrapper that limits reads to a specified number of bytes. + + Enables streaming a slice of a file for upload without loading the + entire slice into memory. The __len__ method lets requests determine + the correct Content-Length for the upload. + """ + + def __init__(self, file_obj, length: int): + self._file = file_obj + self._length = length + + def read(self, size: int = -1) -> bytes: + if size is None or size < 0: + size = self._length + else: + size = min(size, self._length) + data = self._file.read(size) + self._length -= len(data) + return data + + def __len__(self) -> int: + return self._length + + class InvenioRecord(TypedDict): id: str title: str @@ -553,15 +578,14 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): log.debug(f"Uploading part {part_index}: bytes {start_byte}-{end_byte-1} ({part_content_length} bytes)") - # Read the entire part into memory and upload - with open(file_path, "rb") as f: - f.seek(start_byte) - part_data = f.read(part_content_length) - + # Stream the file slice without loading the entire part into memory # Use empty headers - presigned URLs are authenticated via query parameters # Adding Authorization or other headers would invalidate the signature - response = requests.put(part_url, data=part_data) - self._ensure_response_has_expected_status_code(response, 200) + with open(file_path, "rb") as f: + f.seek(start_byte) + reader = _LimitedFileReader(f, part_content_length) + response = requests.put(part_url, data=reader) + self._ensure_response_has_expected_status_code(response, 200) def download_file_from_container( self, From bd71732b7ff1847045c7a2e9706bc8dc6c1652ab Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Tue, 14 Jul 2026 13:59:38 +0200 Subject: [PATCH 08/24] add more test with http mocking to invenio multipart upload --- test/unit/files/test_invenio_multipart.py | 332 ++++++++++++++++++++++ 1 file changed, 332 insertions(+) diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index c37dee2eb7b..a3bcd91f7b7 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -1,9 +1,13 @@ """Unit tests for Invenio multipart upload functionality.""" +from unittest.mock import MagicMock, patch + import pytest from galaxy.files.sources.invenio import ( + _LimitedFileReader, calculate_multipart_params, + InvenioRepositoryInteractor, MAX_UPLOAD_PART_SIZE, MAX_UPLOAD_PARTS, MIN_UPLOAD_PART_SIZE, @@ -116,3 +120,331 @@ class TestCalculateMultipartParams: assert parts <= MAX_UPLOAD_PARTS # Part size should have increased assert part_size > MIN_UPLOAD_PART_SIZE + + +def _mock_response(status_code=200, json_data=None, text=""): + """Create a mock requests.Response object.""" + response = MagicMock() + response.status_code = status_code + response.text = text + if json_data is not None: + response.json.return_value = json_data + return response + + +def _make_interactor(): + """Create an InvenioRepositoryInteractor with a minimal mock plugin.""" + plugin = MagicMock() + plugin.label = "test-invenio" + plugin.get_uri_root.return_value = "inveniordm://test" + return InvenioRepositoryInteractor("https://invenio.example.org", plugin) + + +def _make_context(config=None): + """Create a mock runtime context with the given config.""" + context = MagicMock() + context.config = config or MagicMock( + multipart_threshold=None, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + ) + return context + + +def _make_draft_record(): + """Standard draft record response with files link.""" + return {"links": {"files": "https://invenio.example.org/api/records/abc/files"}} + + +def _make_single_upload_entry(): + """File entry returned after POST for single upload.""" + return { + "key": "test.txt", + "links": { + "content": "https://invenio.example.org/api/records/abc/files/content", + "commit": "https://invenio.example.org/api/records/abc/files/commit", + }, + } + + +def _make_multipart_entries(num_parts): + """File entry returned after POST for multipart upload, with part links.""" + parts = [{"part": i, "url": f"https://invenio.example.org/parts/{i}"} for i in range(num_parts)] + return { + "key": "test.txt", + "links": { + "commit": "https://invenio.example.org/api/records/abc/files/commit", + "self": "https://invenio.example.org/api/records/abc/files/test.txt", + "parts": parts, + }, + } + + +class TestUploadFileSingle: + """Tests for _upload_file_single.""" + + def test_upload_file_single_success(self, tmp_path): + """Verify POST (metadata), PUT (content), POST (commit) sequence.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"hello world") + + interactor = _make_interactor() + context = _make_context() + + with ( + patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()), + patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}), + patch("galaxy.files.sources.invenio.requests") as mock_requests, + ): + mock_requests.post.side_effect = [ + _mock_response(201, {"entries": [_make_single_upload_entry()]}), # metadata + _mock_response(200), # commit + ] + mock_requests.put.return_value = _mock_response(200) + + interactor._upload_file_single("abc", "test.txt", str(file_path), context, file_size=11) + + assert mock_requests.post.call_count == 2 + assert mock_requests.put.call_count == 1 + + def test_upload_file_single_413_raises_helpful_error(self, tmp_path): + """HTTP 413 should raise an error mentioning multipart_threshold.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"hello world") + + interactor = _make_interactor() + context = _make_context() + + with ( + patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()), + patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}), + patch("galaxy.files.sources.invenio.requests") as mock_requests, + ): + mock_requests.post.return_value = _mock_response(201, {"entries": [_make_single_upload_entry()]}) + mock_requests.put.return_value = _mock_response(413) + + with pytest.raises(Exception, match="multipart_threshold"): + interactor._upload_file_single("abc", "test.txt", str(file_path), context, file_size=11) + + +class TestUploadFileMultipart: + """Tests for _upload_file_multipart.""" + + def test_upload_file_multipart_success(self, tmp_path): + """Verify transfer metadata, part uploads, and commit call.""" + file_size = 3 * MIN_UPLOAD_PART_SIZE + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"x" * file_size) + + interactor = _make_interactor() + context = _make_context(config=MagicMock( + multipart_threshold=1, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + )) + + entries = _make_multipart_entries(3) + + with ( + patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()), + patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}), + patch.object(interactor, "_upload_single_part") as mock_upload_part, + patch("galaxy.files.sources.invenio.requests") as mock_requests, + ): + mock_requests.post.side_effect = [ + _mock_response(201, {"entries": [entries]}), # initial POST + _mock_response(200), # commit POST + ] + + interactor._upload_file_multipart("abc", "test.txt", str(file_path), file_size, context) + + # Verify transfer metadata in the initial POST + initial_post_args = mock_requests.post.call_args_list[0] + sent_metadata = initial_post_args.kwargs["json"][0] + assert sent_metadata["key"] == "test.txt" + assert sent_metadata["size"] == file_size + assert sent_metadata["transfer"]["type"] == "M" + assert sent_metadata["transfer"]["parts"] == 3 + assert sent_metadata["transfer"]["part_size"] == MIN_UPLOAD_PART_SIZE + + # All 3 parts uploaded + assert mock_upload_part.call_count == 3 + # Commit called (second POST) + assert mock_requests.post.call_count == 2 + + def test_upload_file_multipart_wrong_part_count_from_server(self, tmp_path): + """Server returning fewer part links than expected should raise.""" + file_size = 3 * MIN_UPLOAD_PART_SIZE + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"x" * file_size) + + interactor = _make_interactor() + context = _make_context(config=MagicMock( + multipart_threshold=1, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + )) + + # Server returns only 2 part links instead of 3 + entries = _make_multipart_entries(2) + + with ( + patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()), + patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}), + patch("galaxy.files.sources.invenio.requests") as mock_requests, + ): + mock_requests.post.return_value = _mock_response(201, {"entries": [entries]}) + + with pytest.raises(Exception, match="2 part URLs"): + interactor._upload_file_multipart("abc", "test.txt", str(file_path), file_size, context) + + +class TestUploadParts: + """Tests for _upload_parts sequential vs parallel behavior.""" + + def test_upload_parts_sequential_for_two_parts(self): + """<=2 parts should upload sequentially without ThreadPoolExecutor.""" + interactor = _make_interactor() + part_links = [{"part": 0, "url": "u0"}, {"part": 1, "url": "u1"}] + + with patch.object(interactor, "_upload_single_part") as mock_upload: + interactor._upload_parts("/path", 1024, 512, part_links, {}) + assert mock_upload.call_count == 2 + + def test_upload_parts_parallel_for_many_parts(self): + """>2 parts should use ThreadPoolExecutor.""" + interactor = _make_interactor() + part_links = [{"part": i, "url": f"u{i}"} for i in range(5)] + + with patch.object(interactor, "_upload_single_part") as mock_upload: + interactor._upload_parts("/path", 5120, 1024, part_links, {}) + assert mock_upload.call_count == 5 + + +class TestUploadSinglePart: + """Tests for _upload_single_part byte-range streaming.""" + + def test_upload_single_part_streams_correct_byte_range(self, tmp_path): + """Verify the uploaded data matches the expected byte slice.""" + data = b"0123456789ABCDEF" * 4 # 64 bytes + file_path = tmp_path / "test.txt" + file_path.write_bytes(data) + + interactor = _make_interactor() + part_size = 16 + part_info = {"part": 2, "url": "https://invenio.example.org/parts/2"} + + captured_data = bytearray() + + def fake_put(url, data=None, **kwargs): + captured_data.extend(data.read()) + return _mock_response(200) + + with patch("galaxy.files.sources.invenio.requests") as mock_requests: + mock_requests.put.side_effect = fake_put + + interactor._upload_single_part(str(file_path), len(data), part_size, 2, part_info) + + # Part 2 = bytes 32-48 + assert bytes(captured_data) == data[32:48] + + +class TestLimitedFileReader: + """Tests for _LimitedFileReader streaming wrapper.""" + + def test_read_limited_bytes(self, tmp_path): + """Reader should only return the specified number of bytes.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"0123456789") + + with open(file_path, "rb") as f: + f.seek(3) + reader = _LimitedFileReader(f, 4) + assert reader.read() == b"3456" + + def test_read_in_chunks(self, tmp_path): + """Reader should serve data in small chunks without over-reading.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"0123456789") + + with open(file_path, "rb") as f: + f.seek(2) + reader = _LimitedFileReader(f, 5) + assert reader.read(2) == b"23" + assert reader.read(2) == b"45" + assert reader.read(2) == b"6" # only 1 byte remaining + assert reader.read(2) == b"" # now exhausted + + def test_len_reports_remaining(self, tmp_path): + """__len__ should report remaining bytes.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"0123456789") + + with open(file_path, "rb") as f: + f.seek(1) + reader = _LimitedFileReader(f, 5) + assert len(reader) == 5 + reader.read(2) + assert len(reader) == 3 + + +class TestUploadFileToDraftContainerRouting: + """Tests for upload_file_to_draft_container threshold routing.""" + + def test_routes_to_single_when_below_threshold(self, tmp_path): + """Files below threshold should use single upload.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"small") + + interactor = _make_interactor() + context = _make_context(config=MagicMock( + multipart_threshold=100, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + )) + + with patch.object(interactor, "_upload_file_single") as mock_single: + interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) + mock_single.assert_called_once() + + def test_routes_to_multipart_when_above_threshold(self, tmp_path): + """Files at or above threshold should use multipart upload.""" + file_size = 2 * 1024 * 1024 # 2 MiB + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"x" * file_size) + + interactor = _make_interactor() + context = _make_context(config=MagicMock( + multipart_threshold=1, # 1 MB threshold, file is 2 MB + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + )) + + with patch.object(interactor, "_upload_file_multipart") as mock_multipart: + interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) + mock_multipart.assert_called_once() + # Verify file_size is passed through + assert mock_multipart.call_args[0][3] == file_size + + def test_routes_to_single_when_no_threshold(self, tmp_path): + """No threshold configured should always use single upload.""" + file_size = 100 * 1024 * 1024 # 100 MiB + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"x" * file_size) + + interactor = _make_interactor() + context = _make_context() + + with patch.object(interactor, "_upload_file_single") as mock_single: + interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) + mock_single.assert_called_once() From bb779994b4139e19decaa8a1643080bddcedeb56 Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Tue, 14 Jul 2026 15:25:37 +0200 Subject: [PATCH 09/24] explicitly check for presigned url and do not use headers for these use headers elsewhere --- lib/galaxy/files/sources/invenio.py | 15 +++++++++++---- test/unit/files/test_invenio_multipart.py | 2 +- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index f99dd5b8f20..2004a9ff177 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -535,7 +535,7 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): if num_parts <= 2: for part_index, part_info in enumerate(part_links): - self._upload_single_part(file_path, file_size, part_size, part_index, part_info) + self._upload_single_part(file_path, file_size, part_size, part_index, part_info, headers) else: max_workers = min(4, num_parts) with ThreadPoolExecutor(max_workers=max_workers) as executor: @@ -548,6 +548,7 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): part_size, part_index, part_info, + headers, ) futures[future] = part_index @@ -566,6 +567,7 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): part_size: int, part_index: int, part_info: dict, + headers: dict, ): """Upload a single part of a multipart upload.""" part_url = part_info.get("url") @@ -578,13 +580,18 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): log.debug(f"Uploading part {part_index}: bytes {start_byte}-{end_byte-1} ({part_content_length} bytes)") + # Presigned S3 URLs are authenticated via query parameters; adding + # Authorization or other headers would invalidate the signature. + # Invenio API proxy URLs require the Authorization header. + # Check for both X-Amz-Signature (current) and Signature (legacy v2) query params. + is_presigned = "X-Amz-Signature" in part_url or "Signature=" in part_url + part_headers = None if is_presigned else headers + # Stream the file slice without loading the entire part into memory - # Use empty headers - presigned URLs are authenticated via query parameters - # Adding Authorization or other headers would invalidate the signature with open(file_path, "rb") as f: f.seek(start_byte) reader = _LimitedFileReader(f, part_content_length) - response = requests.put(part_url, data=reader) + response = requests.put(part_url, data=reader, headers=part_headers) self._ensure_response_has_expected_status_code(response, 200) def download_file_from_container( diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index a3bcd91f7b7..20e8d309219 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -349,7 +349,7 @@ class TestUploadSinglePart: with patch("galaxy.files.sources.invenio.requests") as mock_requests: mock_requests.put.side_effect = fake_put - interactor._upload_single_part(str(file_path), len(data), part_size, 2, part_info) + interactor._upload_single_part(str(file_path), len(data), part_size, 2, part_info, {}) # Part 2 = bytes 32-48 assert bytes(captured_data) == data[32:48] From 6964aebd13d2ba6066e298472ff96151ae36015f Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Tue, 14 Jul 2026 15:27:24 +0200 Subject: [PATCH 10/24] format --- lib/galaxy/files/sources/invenio.py | 4 +- test/unit/files/test_invenio_multipart.py | 69 +++++++++++++---------- 2 files changed, 42 insertions(+), 31 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 2004a9ff177..534ed25cd8f 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -16,12 +16,12 @@ from typing import ( ) from urllib.parse import quote -log = logging.getLogger(__name__) - from typing_extensions import ( TypedDict, ) +log = logging.getLogger(__name__) + from galaxy.exceptions import ( AuthenticationRequired, MessageException, diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index 20e8d309219..3cc7c36632b 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -1,6 +1,9 @@ """Unit tests for Invenio multipart upload functionality.""" -from unittest.mock import MagicMock, patch +from unittest.mock import ( + MagicMock, + patch, +) import pytest @@ -239,13 +242,15 @@ class TestUploadFileMultipart: file_path.write_bytes(b"x" * file_size) interactor = _make_interactor() - context = _make_context(config=MagicMock( - multipart_threshold=1, - multipart_chunk_size=None, - default_resource_type=None, - token="test-token", - public_name="Doe, Jane", - )) + context = _make_context( + config=MagicMock( + multipart_threshold=1, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + ) + ) entries = _make_multipart_entries(3) @@ -283,13 +288,15 @@ class TestUploadFileMultipart: file_path.write_bytes(b"x" * file_size) interactor = _make_interactor() - context = _make_context(config=MagicMock( - multipart_threshold=1, - multipart_chunk_size=None, - default_resource_type=None, - token="test-token", - public_name="Doe, Jane", - )) + context = _make_context( + config=MagicMock( + multipart_threshold=1, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + ) + ) # Server returns only 2 part links instead of 3 entries = _make_multipart_entries(2) @@ -403,13 +410,15 @@ class TestUploadFileToDraftContainerRouting: file_path.write_bytes(b"small") interactor = _make_interactor() - context = _make_context(config=MagicMock( - multipart_threshold=100, - multipart_chunk_size=None, - default_resource_type=None, - token="test-token", - public_name="Doe, Jane", - )) + context = _make_context( + config=MagicMock( + multipart_threshold=100, + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + ) + ) with patch.object(interactor, "_upload_file_single") as mock_single: interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) @@ -422,13 +431,15 @@ class TestUploadFileToDraftContainerRouting: file_path.write_bytes(b"x" * file_size) interactor = _make_interactor() - context = _make_context(config=MagicMock( - multipart_threshold=1, # 1 MB threshold, file is 2 MB - multipart_chunk_size=None, - default_resource_type=None, - token="test-token", - public_name="Doe, Jane", - )) + context = _make_context( + config=MagicMock( + multipart_threshold=1, # 1 MB threshold, file is 2 MB + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + ) + ) with patch.object(interactor, "_upload_file_multipart") as mock_multipart: interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) From 9db007460eb68e2cadd88807199f9b1dc5a1731a Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Tue, 14 Jul 2026 15:30:31 +0200 Subject: [PATCH 11/24] allow user templateing the invenio resource type --- lib/galaxy/files/sources/_rdm.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/files/sources/_rdm.py b/lib/galaxy/files/sources/_rdm.py index e3a142286ac..215e920a1d4 100644 --- a/lib/galaxy/files/sources/_rdm.py +++ b/lib/galaxy/files/sources/_rdm.py @@ -27,7 +27,7 @@ class RDMFileSourceTemplateConfiguration(BaseFileSourceTemplateConfiguration): public_name: Optional[Union[str, TemplateExpansion]] = None multipart_threshold: Optional[Union[int, TemplateExpansion]] = None # MB multipart_chunk_size: Optional[Union[int, TemplateExpansion]] = None # MB - default_resource_type: Optional[str] = None + default_resource_type: Optional[Union[str, TemplateExpansion]] = None class RDMFileSourceConfiguration(BaseFileSourceConfiguration): From 29b200ecdf17144a211b04c3a0f72c54b253174e Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Thu, 16 Jul 2026 14:06:38 +0200 Subject: [PATCH 12/24] only call filesize when needed --- lib/galaxy/files/sources/invenio.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 534ed25cd8f..613faf6476c 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -422,7 +422,7 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): if use_multipart: self._upload_file_multipart(record_id, filename, file_path, file_size, context) else: - self._upload_file_single(record_id, filename, file_path, context, file_size) + self._upload_file_single(record_id, filename, file_path, context) def _upload_file_single( self, @@ -430,11 +430,8 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): filename: str, file_path: str, context: FilesSourceRuntimeContext[RDMFileSourceConfiguration], - file_size: Optional[int] = None, ): """Upload a file using single PUT request.""" - if file_size is None: - file_size = os.path.getsize(file_path) record = self._get_draft_record(record_id, context) upload_file_url = record["links"]["files"] @@ -453,6 +450,7 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): response = requests.put(upload_file_content_url, data=file, headers=headers) # Handle 413 (Payload Too Large) - suggest using multipart upload if response.status_code == 413: + file_size = os.path.getsize(file_path) raise Exception( f"Failed to upload file '{filename}' ({file_size} bytes): HTTP 413 Payload Too Large. " f"The server rejected the upload because the file is too large for a single request. " From b6391975c6a3cfbaa59025be51148a288346177e Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Thu, 16 Jul 2026 14:19:57 +0200 Subject: [PATCH 13/24] refactor 413 handling up to the router --- lib/galaxy/files/sources/invenio.py | 19 ++++++------- test/unit/files/test_invenio_multipart.py | 33 ++++++++++++++++++++--- 2 files changed, 40 insertions(+), 12 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 613faf6476c..1b24e2158eb 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -422,7 +422,16 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): if use_multipart: self._upload_file_multipart(record_id, filename, file_path, file_size, context) else: - self._upload_file_single(record_id, filename, file_path, context) + try: + self._upload_file_single(record_id, filename, file_path, context) + except Exception as e: + if "413" in str(e): + raise Exception( + f"Failed to upload file '{filename}' ({file_size} bytes): HTTP 413 Payload Too Large. " + f"The server rejected the upload because the file is too large for a single request. " + f"Please configure 'multipart_threshold' in the file source configuration to enable multipart upload for files of this size." + ) from e + raise def _upload_file_single( self, @@ -448,14 +457,6 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): commit_file_upload_url = file_entry["links"]["commit"] with open(file_path, "rb") as file: response = requests.put(upload_file_content_url, data=file, headers=headers) - # Handle 413 (Payload Too Large) - suggest using multipart upload - if response.status_code == 413: - file_size = os.path.getsize(file_path) - raise Exception( - f"Failed to upload file '{filename}' ({file_size} bytes): HTTP 413 Payload Too Large. " - f"The server rejected the upload because the file is too large for a single request. " - f"Please configure 'multipart_threshold' in the file source configuration to enable multipart upload for files of this size." - ) self._ensure_response_has_expected_status_code(response, 200) # Commit file upload diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index 3cc7c36632b..c2bb82f7a73 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -207,7 +207,7 @@ class TestUploadFileSingle: ] mock_requests.put.return_value = _mock_response(200) - interactor._upload_file_single("abc", "test.txt", str(file_path), context, file_size=11) + interactor._upload_file_single("abc", "test.txt", str(file_path), context) assert mock_requests.post.call_count == 2 assert mock_requests.put.call_count == 1 @@ -228,8 +228,8 @@ class TestUploadFileSingle: mock_requests.post.return_value = _mock_response(201, {"entries": [_make_single_upload_entry()]}) mock_requests.put.return_value = _mock_response(413) - with pytest.raises(Exception, match="multipart_threshold"): - interactor._upload_file_single("abc", "test.txt", str(file_path), context, file_size=11) + with pytest.raises(Exception, match="413"): + interactor._upload_file_single("abc", "test.txt", str(file_path), context) class TestUploadFileMultipart: @@ -459,3 +459,30 @@ class TestUploadFileToDraftContainerRouting: with patch.object(interactor, "_upload_file_single") as mock_single: interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) mock_single.assert_called_once() + + def test_upload_file_single_413_router_raises_helpful_error(self, tmp_path): + """Router should wrap 413 with actionable multipart_threshold message.""" + file_path = tmp_path / "test.txt" + file_path.write_bytes(b"hello world") + + interactor = _make_interactor() + context = _make_context( + config=MagicMock( + multipart_threshold=None, # No threshold configured + multipart_chunk_size=None, + default_resource_type=None, + token="test-token", + public_name="Doe, Jane", + ) + ) + + with ( + patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()), + patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}), + patch("galaxy.files.sources.invenio.requests") as mock_requests, + ): + mock_requests.post.return_value = _mock_response(201, {"entries": [_make_single_upload_entry()]}) + mock_requests.put.return_value = _mock_response(413) + + with pytest.raises(Exception, match="multipart_threshold"): + interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context) From f556211855c06644d95186fcf1d96f0834b709fe Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Thu, 16 Jul 2026 14:25:33 +0200 Subject: [PATCH 14/24] remove the branching for a single part, drop superficial tests --- lib/galaxy/files/sources/invenio.py | 46 +++++++++++------------ test/unit/files/test_invenio_multipart.py | 22 ----------- 2 files changed, 21 insertions(+), 47 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 1b24e2158eb..cabbfa34087 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -529,35 +529,31 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): part_links: list[dict], headers: dict, ): - """Upload all parts, sequentially for <=2 parts, parallel otherwise.""" + """Upload all parts in parallel using a thread pool.""" num_parts = len(part_links) + max_workers = min(4, num_parts) - if num_parts <= 2: + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = {} for part_index, part_info in enumerate(part_links): - self._upload_single_part(file_path, file_size, part_size, part_index, part_info, headers) - else: - max_workers = min(4, num_parts) - with ThreadPoolExecutor(max_workers=max_workers) as executor: - futures = {} - for part_index, part_info in enumerate(part_links): - future = executor.submit( - self._upload_single_part, - file_path, - file_size, - part_size, - part_index, - part_info, - headers, - ) - futures[future] = part_index + future = executor.submit( + self._upload_single_part, + file_path, + file_size, + part_size, + part_index, + part_info, + headers, + ) + futures[future] = part_index - for future in as_completed(futures): - part_index = futures[future] - try: - future.result() - except Exception as e: - log.error(f"Failed to upload part {part_index}: {e}") - raise + for future in as_completed(futures): + part_index = futures[future] + try: + future.result() + except Exception as e: + log.error(f"Failed to upload part {part_index}: {e}") + raise def _upload_single_part( self, diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index c2bb82f7a73..38062c629d2 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -312,28 +312,6 @@ class TestUploadFileMultipart: interactor._upload_file_multipart("abc", "test.txt", str(file_path), file_size, context) -class TestUploadParts: - """Tests for _upload_parts sequential vs parallel behavior.""" - - def test_upload_parts_sequential_for_two_parts(self): - """<=2 parts should upload sequentially without ThreadPoolExecutor.""" - interactor = _make_interactor() - part_links = [{"part": 0, "url": "u0"}, {"part": 1, "url": "u1"}] - - with patch.object(interactor, "_upload_single_part") as mock_upload: - interactor._upload_parts("/path", 1024, 512, part_links, {}) - assert mock_upload.call_count == 2 - - def test_upload_parts_parallel_for_many_parts(self): - """>2 parts should use ThreadPoolExecutor.""" - interactor = _make_interactor() - part_links = [{"part": i, "url": f"u{i}"} for i in range(5)] - - with patch.object(interactor, "_upload_single_part") as mock_upload: - interactor._upload_parts("/path", 5120, 1024, part_links, {}) - assert mock_upload.call_count == 5 - - class TestUploadSinglePart: """Tests for _upload_single_part byte-range streaming.""" From 766d0812a330d7c8536f60fa85507fa8c2f69493 Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Thu, 16 Jul 2026 14:35:04 +0200 Subject: [PATCH 15/24] set minimal part size --- lib/galaxy/files/sources/invenio.py | 2 +- test/unit/files/test_invenio_multipart.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index cabbfa34087..9c06b14b651 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -132,7 +132,7 @@ def calculate_multipart_params(file_size: int, preferred_part_size: int | None = Files larger than this will still return valid params but would fail server-side. """ if file_size == 0: - return 1, 0 + return 1, MIN_UPLOAD_PART_SIZE # Start with preferred or minimum part size part_size = preferred_part_size or MIN_UPLOAD_PART_SIZE diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index 38062c629d2..b5912ba7088 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -21,10 +21,10 @@ class TestCalculateMultipartParams: """Tests for calculate_multipart_params function.""" def test_calculate_multipart_params_zero_byte(self): - """Zero-byte files should return (1, 0).""" + """Zero-byte files return minimum part size.""" parts, part_size = calculate_multipart_params(0) assert parts == 1 - assert part_size == 0 + assert part_size == MIN_UPLOAD_PART_SIZE def test_calculate_multipart_params_small_file(self): """Files under 5 MiB should use minimum part size.""" From 5ec958fe3f6eb2395bcab3980dd7d2dfb2118c8b Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Thu, 16 Jul 2026 14:40:22 +0200 Subject: [PATCH 16/24] lint --- lib/galaxy/files/sources/invenio.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 9c06b14b651..1a877305f64 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -573,7 +573,7 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): end_byte = min(start_byte + part_size, file_size) part_content_length = end_byte - start_byte - log.debug(f"Uploading part {part_index}: bytes {start_byte}-{end_byte-1} ({part_content_length} bytes)") + log.debug(f"Uploading part {part_index}: bytes {start_byte}-{end_byte - 1} ({part_content_length} bytes)") # Presigned S3 URLs are authenticated via query parameters; adding # Authorization or other headers would invalidate the signature. From 39581244275c2d508ae76eed6a1b968893ebfcab Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Fri, 17 Jul 2026 12:04:40 +0200 Subject: [PATCH 17/24] do not rely on string parsing for exception --- lib/galaxy/files/sources/invenio.py | 17 +++++++++++++---- test/unit/files/test_invenio_multipart.py | 9 ++++++--- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 1a877305f64..8b6be4e7540 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -183,6 +183,14 @@ class _LimitedFileReader: return self._length +class InvenioRequestError(Exception): + """Raised when an Invenio API request returns an unexpected status code.""" + + def __init__(self, message: str, status_code: int): + super().__init__(message) + self.status_code = status_code + + class InvenioRecord(TypedDict): id: str title: str @@ -424,8 +432,8 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): else: try: self._upload_file_single(record_id, filename, file_path, context) - except Exception as e: - if "413" in str(e): + except InvenioRequestError as e: + if e.status_code == 413: raise Exception( f"Failed to upload file '{filename}' ({file_size} bytes): HTTP 413 Payload Too Large. " f"The server rejected the upload because the file is too large for a single request. " @@ -789,8 +797,9 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor): if response.status_code == 403: self._raise_auth_required() error_message = self._get_response_error_message(response) - raise Exception( - f"Request to {response.url} failed with status code {response.status_code}: {error_message}" + raise InvenioRequestError( + f"Request to {response.url} failed with status code {response.status_code}: {error_message}", + status_code=response.status_code, ) def _raise_auth_required(self): diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index b5912ba7088..3d173fba49a 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -11,6 +11,7 @@ from galaxy.files.sources.invenio import ( _LimitedFileReader, calculate_multipart_params, InvenioRepositoryInteractor, + InvenioRequestError, MAX_UPLOAD_PART_SIZE, MAX_UPLOAD_PARTS, MIN_UPLOAD_PART_SIZE, @@ -212,8 +213,8 @@ class TestUploadFileSingle: assert mock_requests.post.call_count == 2 assert mock_requests.put.call_count == 1 - def test_upload_file_single_413_raises_helpful_error(self, tmp_path): - """HTTP 413 should raise an error mentioning multipart_threshold.""" + def test_upload_file_single_propagates_413_as_typed_error(self, tmp_path): + """A 413 from the content PUT surfaces as an InvenioRequestError carrying the status code.""" file_path = tmp_path / "test.txt" file_path.write_bytes(b"hello world") @@ -228,9 +229,11 @@ class TestUploadFileSingle: mock_requests.post.return_value = _mock_response(201, {"entries": [_make_single_upload_entry()]}) mock_requests.put.return_value = _mock_response(413) - with pytest.raises(Exception, match="413"): + with pytest.raises(InvenioRequestError) as exc_info: interactor._upload_file_single("abc", "test.txt", str(file_path), context) + assert exc_info.value.status_code == 413 + class TestUploadFileMultipart: """Tests for _upload_file_multipart.""" From c4e3fae3f8cc69122888e7a2f3fbd6cbdbe0943e Mon Sep 17 00:00:00 2001 From: Martin Cech Date: Fri, 17 Jul 2026 13:00:28 +0200 Subject: [PATCH 18/24] simplify calculating number of parts, raise error when over limit --- lib/galaxy/files/sources/invenio.py | 36 +++++++++++------------ test/unit/files/test_invenio_multipart.py | 23 ++++++++------- 2 files changed, 31 insertions(+), 28 deletions(-) diff --git a/lib/galaxy/files/sources/invenio.py b/lib/galaxy/files/sources/invenio.py index 8b6be4e7540..3d57910bdde 100644 --- a/lib/galaxy/files/sources/invenio.py +++ b/lib/galaxy/files/sources/invenio.py @@ -118,7 +118,7 @@ MAX_UPLOAD_PARTS = 10_000 def calculate_multipart_params(file_size: int, preferred_part_size: int | None = None) -> tuple[int, int]: - """Calculate optimal parts count and part size for multipart upload. + """Calculate parts count and part size for multipart upload. Args: file_size: Total file size in bytes @@ -127,34 +127,34 @@ def calculate_multipart_params(file_size: int, preferred_part_size: int | None = Returns: Tuple of (parts_count, part_size) + Raises: + ValueError: If the file is larger than MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE + (~48.8 TiB), which exceeds the maximum uploadable size. + Note: Maximum uploadable file size is MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB). - Files larger than this will still return valid params but would fail server-side. + Files larger than this cannot be uploaded via multipart and raise ValueError. """ if file_size == 0: return 1, MIN_UPLOAD_PART_SIZE - # Start with preferred or minimum part size + # Start with preferred or minimum part size, clamped to [min, max] part_size = preferred_part_size or MIN_UPLOAD_PART_SIZE - - # Ensure part_size is within bounds part_size = max(part_size, MIN_UPLOAD_PART_SIZE) part_size = min(part_size, MAX_UPLOAD_PART_SIZE) - # Calculate parts needed + # Grow part_size to the minimum that keeps the part count within MAX_UPLOAD_PARTS. + part_size = max(part_size, math.ceil(file_size / MAX_UPLOAD_PARTS)) + part_size = min(part_size, MAX_UPLOAD_PART_SIZE) + + max_upload_size = MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE + if file_size > max_upload_size: + raise ValueError( + f"File size {file_size} bytes exceeds the maximum multipart upload size " + f"of {max_upload_size} bytes ({MAX_UPLOAD_PARTS} parts x {MAX_UPLOAD_PART_SIZE} bytes)." + ) + parts = math.ceil(file_size / part_size) - - # If too many parts, increase part size (up to max) - while parts > MAX_UPLOAD_PARTS and part_size < MAX_UPLOAD_PART_SIZE: - part_size = min(part_size * 2, MAX_UPLOAD_PART_SIZE) - parts = math.ceil(file_size / part_size) - - # For extremely large files, cap parts at MAX_UPLOAD_PARTS - # This means part_size may effectively be larger than calculated - # but such files would likely fail server-side anyway - if parts > MAX_UPLOAD_PARTS: - parts = MAX_UPLOAD_PARTS - return parts, part_size diff --git a/test/unit/files/test_invenio_multipart.py b/test/unit/files/test_invenio_multipart.py index 3d173fba49a..c1e3f7b2896 100644 --- a/test/unit/files/test_invenio_multipart.py +++ b/test/unit/files/test_invenio_multipart.py @@ -60,19 +60,22 @@ class TestCalculateMultipartParams: assert part_size >= MIN_UPLOAD_PART_SIZE def test_calculate_multipart_params_extremely_large_file(self): - """Extremely large files should hit both MAX limits. + """Files larger than the maximum uploadable size raise ValueError instead of truncating.""" + # 100 TiB file - exceeds theoretical s3 maximum (~48.8 TiB) + file_size = 100 * 1024**4 + with pytest.raises(ValueError, match="exceeds the maximum multipart upload size"): + calculate_multipart_params(file_size) - Note: Files larger than MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB) - cannot be uploaded via multipart, but we cap params rather than fail here. - The upload would fail server-side anyway. - """ - # 100 TiB file - exceeds theoretical maximum (~48.8 TiB) - file_size = 100 * 1024**4 # 100 TiB - parts, part_size = calculate_multipart_params(file_size) - # Parts should be capped at MAX_UPLOAD_PARTS + def test_calculate_multipart_params_raises_at_max_boundary(self): + """One byte over the maximum uploadable size raises; exactly at the max does not.""" + max_upload_size = MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE + # Exactly at the boundary is allowed (parts == MAX_UPLOAD_PARTS at MAX part size). + parts, part_size = calculate_multipart_params(max_upload_size) assert parts == MAX_UPLOAD_PARTS - # Part size should hit max assert part_size == MAX_UPLOAD_PART_SIZE + # One byte over raises. + with pytest.raises(ValueError): + calculate_multipart_params(max_upload_size + 1) def test_calculate_multipart_params_respects_preferred_part_size(self): """Should use preferred part size when provided and valid.""" From a7f8b9a0d3169d5344e291826804bf13b369d4a1 Mon Sep 17 00:00:00 2001 From: Alireza Heidari <8046843+itisAliRH@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:37:46 +0200 Subject: [PATCH 19/24] fix: align workflow preview layout --- .../Workflow/Published/WorkflowPublished.vue | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/client/src/components/Workflow/Published/WorkflowPublished.vue b/client/src/components/Workflow/Published/WorkflowPublished.vue index 3f7553ba3a4..94ee24488e1 100644 --- a/client/src/components/Workflow/Published/WorkflowPublished.vue +++ b/client/src/components/Workflow/Published/WorkflowPublished.vue @@ -177,34 +177,47 @@ defineExpose({ .container-root { container-type: inline-size; width: 100%; + height: 100%; + min-height: 0; overflow: auto; } .published-workflow { display: grid; gap: 0.5rem 1rem; - grid-template-columns: auto auto 30%; + grid-template-columns: minmax(0, 1fr) minmax(18rem, 30%); + grid-template-rows: auto minmax(0, 1fr); height: 100%; + min-height: 0; .workflow-header { - grid-column: 1 / span 3; + grid-column: 1 / -1; display: flex; + align-items: center; + gap: 1rem; justify-content: flex-end; } .workflow-preview { - grid-column: 1 / span 2; + grid-column: 1; + min-height: 0; &.only-preview { - grid-column: 1 / span 3; + grid-column: 1 / -1; + } + + &:deep(.card-body) { + height: 100%; + min-height: 0; } } &:deep(.workflow-information-container) { height: 100%; max-width: 500px; + align-self: stretch; overflow: auto; } } @@ -212,10 +225,11 @@ defineExpose({ @container (max-width: 900px) { .published-workflow { height: unset; - grid-template-columns: auto; + grid-template-columns: minmax(0, 1fr); + grid-template-rows: auto auto auto; .workflow-preview { - grid-column: 1 / span 3; + grid-column: 1; height: 450px; } @@ -225,7 +239,7 @@ defineExpose({ } .workflow-information-container { - grid-column: 1 / span 3; + grid-column: 1; } } } From e4e29c74e1824f1e201a540cdcb680efa527fb4f Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 21 Jul 2026 08:16:48 -0400 Subject: [PATCH 20/24] WES: use standard workflow accessibility check for gxworkflow:// refs Ownership-only check rejected published/shared workflows that a normal invocation would accept. Defer to get_stored_accessible_workflow, update docs, add API tests for published + inaccessible cases. Co-Authored-By: Claude Opus 4.8 (1M context) --- doc/source/dev/ga4gh_wes.md | 28 +++++++++-------- lib/galaxy/webapps/galaxy/services/wes.py | 18 +++-------- lib/galaxy_test/api/test_wes.py | 38 +++++++++++++++++++++++ 3 files changed, 58 insertions(+), 26 deletions(-) diff --git a/doc/source/dev/ga4gh_wes.md b/doc/source/dev/ga4gh_wes.md index 5556de4281a..bd26d26e698 100644 --- a/doc/source/dev/ga4gh_wes.md +++ b/doc/source/dev/ga4gh_wes.md @@ -129,16 +129,16 @@ Trimmed to the WES-relevant fields (the real response also carries A WES `RunRequest` is `multipart/form-data`. The fields Galaxy honors: -| Field | Required | Notes | -| --------------------------------------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------- | -| `workflow_type` | yes | `gx_workflow_format2` or `gx_workflow_ga`. Must match the auto-detected type or you get a 400. | -| `workflow_type_version` | yes | Free-form string, e.g. `"1.0.0"`. | -| `workflow_url` | one of url/attachment | A URL Galaxy can fetch (`http(s)`, `s3`, `gs`, `file`, `base64://`) **or** a `gxworkflow://` reference (see below). | -| `workflow_attachment` | one of url/attachment | The workflow file uploaded inline. | -| `workflow_params` | no | JSON object of workflow inputs (see below). | -| `workflow_engine_parameters` | no | JSON object of Galaxy-specific run options (see below). | -| `tags` | no | Accepted but currently not persisted onto the invocation. | -| `workflow_engine` / `workflow_engine_version` | no | Accepted; informational. | +| Field | Required | Notes | +| --------------------------------------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------------- | +| `workflow_type` | yes | `gx_workflow_format2` or `gx_workflow_ga`. Must match the auto-detected type or you get a 400. | +| `workflow_type_version` | yes | Free-form string, e.g. `"1.0.0"`. | +| `workflow_url` | one of url/attachment | A URL Galaxy can fetch (`http(s)://`, `s3://`, `gs://`, `file://`, `base64://`) **or** a `gxworkflow://` reference (see below). | +| `workflow_attachment` | one of url/attachment | The workflow file uploaded inline. | +| `workflow_params` | no | JSON object of workflow inputs (see below). | +| `workflow_engine_parameters` | no | JSON object of Galaxy-specific run options (see below). | +| `tags` | no | Accepted but currently not persisted onto the invocation. | +| `workflow_engine` / `workflow_engine_version` | no | Accepted; informational. | ### `workflow_params` — wiring up inputs @@ -180,7 +180,9 @@ gxworkflow:// # the StoredWorkflow (latest versi gxworkflow://?instance=true # a specific Workflow instance ``` -The caller must own the workflow (or be an admin). With `gxworkflow://`, Galaxy skips +The workflow just has to be accessible to the caller — the same rule a normal invocation +uses: owned by them, shared with them, published/importable, or the caller is an admin. +Otherwise you get a 403. With `gxworkflow://`, Galaxy skips import and invokes the stored workflow directly. `workflow_type` is still required by the form but is **not** validated against the stored workflow in this case — the "must match the auto-detected type or 400" check only applies to inline @@ -400,8 +402,8 @@ Access to resources you do not own is **not** reported uniformly — watch for t - Reading a **run** you don't own → `403` (`AuthenticationRequired`). - Submitting against a **history** you don't own → `404` (`ObjectNotFound`). -- A `gxworkflow://` reference to a **workflow** you don't own → `403` - (`ItemAccessibilityException`). +- A `gxworkflow://` reference to a **workflow** you cannot access (not owned, not shared + with you, not published) → `403` (`ItemAccessibilityException`). So a `404` on submission can mean "your history id is wrong" _or_ "that history belongs to someone else"; don't assume `404` always means the object is absent. diff --git a/lib/galaxy/webapps/galaxy/services/wes.py b/lib/galaxy/webapps/galaxy/services/wes.py index 8ed1a839275..3d3ac701847 100644 --- a/lib/galaxy/webapps/galaxy/services/wes.py +++ b/lib/galaxy/webapps/galaxy/services/wes.py @@ -483,22 +483,14 @@ class WesService(ServiceBase): workflow_uri = workflow_dict["workflow_uri"] encoded_workflow_id, instance = _parse_gxworkflow_uri(workflow_uri) - # Load the workflow from the database + # Load the workflow from the database, applying the same accessibility + # rules as a normal invocation (owned, shared, published, or admin). # by_stored_id=not instance means: # - False (instance=False) -> load StoredWorkflow (by_stored_id=True) # - True (instance=True) -> load Workflow (by_stored_id=False) - try: - stored_workflow = self._workflows_service._workflows_manager.get_stored_workflow( - trans, encoded_workflow_id, by_stored_id=not instance - ) - except Exception as e: - raise exceptions.ObjectNotFound( - f"Workflow '{encoded_workflow_id}' not found or not accessible: {str(e)}" - ) - - # Validate user has access to this workflow - if stored_workflow.user_id != trans.user.id and not trans.user_is_admin: - raise exceptions.ItemAccessibilityException("You do not have access to this workflow") + stored_workflow = self._workflows_service._workflows_manager.get_stored_accessible_workflow( + trans, encoded_workflow_id, by_stored_id=not instance + ) # Use the existing workflow directly - no need to create a new one # Skip to step 5 (engine parameters and history) diff --git a/lib/galaxy_test/api/test_wes.py b/lib/galaxy_test/api/test_wes.py index 1b601595547..4eba1a99748 100644 --- a/lib/galaxy_test/api/test_wes.py +++ b/lib/galaxy_test/api/test_wes.py @@ -546,6 +546,44 @@ steps: # Validate response assert run_id is not None + def test_wes_submit_run_with_gxworkflow_uri_published_workflow(self): + """A published workflow owned by another user can be run via gxworkflow://.""" + with self._different_user(): + workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE) + self.workflow_populator.make_public(workflow_id) + + with self.dataset_populator.test_history() as history_id: + dataset_id = self._get_test_dataset_id(history_id) + + data = { + "workflow_type": "gx_workflow_ga", + "workflow_type_version": "v1", + "workflow_params": json.dumps({"input1": dataset_id}), + "workflow_url": f"gxworkflow://{workflow_id}", + } + + response = self._wes_post("ga4gh/wes/v1/runs", data=data) + self._assert_status_code_is(response, 200) + assert response.json()["run_id"] is not None + + def test_wes_submit_run_with_gxworkflow_uri_inaccessible_workflow(self): + """An unshared workflow owned by another user cannot be run via gxworkflow://.""" + with self._different_user(): + workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE) + + with self.dataset_populator.test_history() as history_id: + dataset_id = self._get_test_dataset_id(history_id) + + data = { + "workflow_type": "gx_workflow_ga", + "workflow_type_version": "v1", + "workflow_params": json.dumps({"input1": dataset_id}), + "workflow_url": f"gxworkflow://{workflow_id}", + } + + response = self._wes_post("ga4gh/wes/v1/runs", data=data) + self._assert_status_code_is(response, 403) + def test_wes_job_stdout_endpoint(self): """Test /api/jobs/{job_id}/stdout endpoint returns job stdout.""" with self.dataset_populator.test_history() as history_id: From 90e17de3fe85239226dcc83e9b4ef22d4b8e1c54 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 21 Jul 2026 08:26:06 -0400 Subject: [PATCH 21/24] WES: gxworkflow://?instance=true ran latest version, not referenced one Resolved the Workflow instance for the access check but then invoked by stored workflow id with instance unset. Pass the instance id and flag through to invoke_workflow; strengthen the test to assert the invocation used the referenced version. Co-Authored-By: Claude Opus 4.8 (1M context) --- lib/galaxy/webapps/galaxy/services/wes.py | 13 ++++++++++--- lib/galaxy_test/api/test_wes.py | 15 ++++++++++++--- 2 files changed, 22 insertions(+), 6 deletions(-) diff --git a/lib/galaxy/webapps/galaxy/services/wes.py b/lib/galaxy/webapps/galaxy/services/wes.py index 3d3ac701847..02837ed8702 100644 --- a/lib/galaxy/webapps/galaxy/services/wes.py +++ b/lib/galaxy/webapps/galaxy/services/wes.py @@ -493,7 +493,11 @@ class WesService(ServiceBase): ) # Use the existing workflow directly - no need to create a new one - # Skip to step 5 (engine parameters and history) + if instance: + # The URI named a specific version - invoke that one, not the latest. + invoke_workflow_id = trans.security.decode_id(encoded_workflow_id) + else: + invoke_workflow_id = stored_workflow.id else: # Step 2: Determine/validate workflow type detected_type = _determine_workflow_type(workflow_dict) @@ -520,6 +524,8 @@ class WesService(ServiceBase): source="WES API", ) stored_workflow = created_workflow.stored_workflow + invoke_workflow_id = stored_workflow.id + instance = False # Step 5: Parse engine parameters and create/select history engine_params = {} @@ -532,9 +538,10 @@ class WesService(ServiceBase): history = _get_or_create_history(trans, engine_params) # Step 6: Parse workflow parameters - invoke_params = { + invoke_params: dict[str, Any] = { "history_id": trans.security.encode_id(history.id), "inputs_by": "name", + "instance": instance, } if workflow_params: @@ -554,7 +561,7 @@ class WesService(ServiceBase): invoke_payload = InvokeWorkflowPayload(**invoke_params) workflow_invocation_response = self._workflows_service.invoke_workflow( trans, - trans.security.encode_id(stored_workflow.id), + invoke_workflow_id, invoke_payload, ) diff --git a/lib/galaxy_test/api/test_wes.py b/lib/galaxy_test/api/test_wes.py index 4eba1a99748..af35bc7651d 100644 --- a/lib/galaxy_test/api/test_wes.py +++ b/lib/galaxy_test/api/test_wes.py @@ -520,16 +520,23 @@ steps: self._assert_status_code_is(response, 200) def test_wes_submit_run_with_gxworkflow_uri_with_instance_param(self): - """Test gxworkflow:// URI with instance=true parameter.""" + """Test gxworkflow:// URI with instance=true runs that version, not the latest.""" with self.dataset_populator.test_history() as history_id: dataset_id = self._get_test_dataset_id(history_id) # Upload a workflow to get its ID workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE) - latest_instance_id = self._latest_instance_id(workflow_id, history_id) + first_instance_id = self._latest_instance_id(workflow_id, history_id) + + # Create a second version so the referenced instance is no longer the latest + workflow_object = self._download_workflow(workflow_id) + workflow_object["steps"]["1"]["annotation"] = "second version" + update_response = self.workflow_populator.update_workflow(workflow_id, workflow_object) + self._assert_status_code_is(update_response, 200) + assert self._latest_instance_id(workflow_id, history_id) != first_instance_id # Construct gxworkflow:// URI with instance=true - workflow_uri = f"gxworkflow://{latest_instance_id}?instance=true" + workflow_uri = f"gxworkflow://{first_instance_id}?instance=true" # Submit workflow using the gxworkflow:// URI data = { @@ -545,6 +552,8 @@ steps: # Validate response assert run_id is not None + invocation = self.workflow_populator.get_invocation(run_id) + assert invocation["workflow_id"] == first_instance_id def test_wes_submit_run_with_gxworkflow_uri_published_workflow(self): """A published workflow owned by another user can be run via gxworkflow://.""" From 250c71cfcc4a41622a2f2bdc77b11d2911691d57 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 21 Jul 2026 09:50:19 -0400 Subject: [PATCH 22/24] WES docs: tighten gxworkflow accessibility wording --- doc/source/dev/ga4gh_wes.md | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/doc/source/dev/ga4gh_wes.md b/doc/source/dev/ga4gh_wes.md index bd26d26e698..1b527e894a4 100644 --- a/doc/source/dev/ga4gh_wes.md +++ b/doc/source/dev/ga4gh_wes.md @@ -180,13 +180,14 @@ gxworkflow:// # the StoredWorkflow (latest versi gxworkflow://?instance=true # a specific Workflow instance ``` -The workflow just has to be accessible to the caller — the same rule a normal invocation +The workflow only has to be accessible to the caller — the same rule a normal invocation uses: owned by them, shared with them, published/importable, or the caller is an admin. -Otherwise you get a 403. With `gxworkflow://`, Galaxy skips -import and invokes the stored workflow directly. `workflow_type` is still required by the -form but is **not** validated against the stored workflow in this case — the -"must match the auto-detected type or 400" check only applies to inline -`workflow_attachment` / fetched `workflow_url` submissions. +Anything else is a 403. + +With `gxworkflow://`, Galaxy skips the import step and invokes the referenced workflow +directly. `workflow_type` is still required by the form but is **not** validated against +the stored workflow in this case — the "must match the auto-detected type or 400" check +only applies to inline `workflow_attachment` / fetched `workflow_url` submissions. ## 3. Submit the run From 36d746941348ab42f87eb39c58a1edd40d7c0086 Mon Sep 17 00:00:00 2001 From: Ahmed Awan Date: Tue, 21 Jul 2026 12:27:37 -0500 Subject: [PATCH 23/24] fix set_history committing None galaxy_session for API-key requests Requests authenticated via API key have no galaxy_session, but SessionRequestContext.set_history unconditionally called sa_session.add/commit on it, which fails when galaxy_session is None. Guard the add/commit behind a galaxy_session check so set_history is a no-op when there's no session to persist. Fixes https://github.com/galaxyproject/galaxy/issues/23148 Add regression tests covering both the missing-session and present-session paths. Co-Authored-By: Claude --- lib/galaxy/work/context.py | 9 ++++--- test/unit/app/test_work_context.py | 39 ++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) create mode 100644 test/unit/app/test_work_context.py diff --git a/lib/galaxy/work/context.py b/lib/galaxy/work/context.py index d1a5831ddfc..f0c1fe6471b 100644 --- a/lib/galaxy/work/context.py +++ b/lib/galaxy/work/context.py @@ -180,10 +180,11 @@ class SessionRequestContext(WorkRequestContext): return self.galaxy_session def set_history(self, history): - if history and not history.deleted and self.galaxy_session: - self.galaxy_session.current_history = history - self.sa_session.add(self.galaxy_session) - self.sa_session.commit() + if self.galaxy_session: + if history and not history.deleted: + self.galaxy_session.current_history = history + self.sa_session.add(self.galaxy_session) + self.sa_session.commit() def proxy_work_context_for_history( diff --git a/test/unit/app/test_work_context.py b/test/unit/app/test_work_context.py new file mode 100644 index 00000000000..99064f07b85 --- /dev/null +++ b/test/unit/app/test_work_context.py @@ -0,0 +1,39 @@ +from unittest import mock + +from galaxy.work.context import SessionRequestContext + + +def _make_context(galaxy_session=None): + return SessionRequestContext( + app=mock.MagicMock(), + request=mock.MagicMock(), + response=mock.MagicMock(), + galaxy_session=galaxy_session, + ) + + +def test_set_history_without_galaxy_session_does_not_raise(): + """Regression test for https://github.com/galaxyproject/galaxy/issues/23148. + + API requests authenticated via API key have no galaxy_session, so set_history + must not unconditionally add/commit a None session to the sa_session. + """ + trans = _make_context(galaxy_session=None) + history = mock.MagicMock(deleted=False) + + trans.set_history(history) + + trans.sa_session.add.assert_not_called() + trans.sa_session.commit.assert_not_called() + + +def test_set_history_with_galaxy_session_updates_and_commits(): + galaxy_session = mock.MagicMock() + trans = _make_context(galaxy_session=galaxy_session) + history = mock.MagicMock(deleted=False) + + trans.set_history(history) + + assert galaxy_session.current_history == history + trans.sa_session.add.assert_called_once_with(galaxy_session) + trans.sa_session.commit.assert_called_once() From 3240ca17f4deb66a7e824808955f6d3d01d6a480 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Wed, 22 Jul 2026 11:26:42 +0200 Subject: [PATCH 24/24] Deny cookieless anonymous history copies instead of orphaning them POST /api/histories may copy a published history for anonymous users, but a request with neither an API key nor a galaxysession cookie has no galaxy_session to own the result. Previously this crashed in set_history (sa_session.add(None), issue #23148); with the set_history guard from the previous commit it would instead silently perform the (potentially expensive) copy and return a history the caller can never reach again. Reject that case up front in HistoriesService.create with a 403 rather than doing wasted work. Legitimate anonymous copies from the browser are unaffected: the SPA shell is served by the legacy WSGI route, which mints a galaxysession cookie, so those requests always carry a session. Replace the set_history unit test with an API test that drives the real request path: a cookieless copy of a published history now returns 403. Co-Authored-By: Claude --- .../webapps/galaxy/services/histories.py | 11 +++++- lib/galaxy_test/api/test_histories.py | 19 ++++++++- test/unit/app/test_work_context.py | 39 ------------------- 3 files changed, 27 insertions(+), 42 deletions(-) delete mode 100644 test/unit/app/test_work_context.py diff --git a/lib/galaxy/webapps/galaxy/services/histories.py b/lib/galaxy/webapps/galaxy/services/histories.py index 36cb84b4ff6..2d8f6b6e0b3 100644 --- a/lib/galaxy/webapps/galaxy/services/histories.py +++ b/lib/galaxy/webapps/galaxy/services/histories.py @@ -262,8 +262,15 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores): from URL or File depending on the provided parameters in the payload. """ copy_this_history_id = payload.history_id - if trans.anonymous and not copy_this_history_id: # Copying/Importing histories is allowed for anonymous users - raise glx_exceptions.AuthenticationRequired("You need to be logged in to create histories.") + if trans.anonymous: + # Copying/Importing histories is allowed for anonymous users, but only when there is a + # galaxy_session to own the result. A request with neither an API key nor a galaxysession + # cookie (e.g. a raw API call) has nowhere to put the new history, so reject it instead of + # doing the (potentially expensive) copy and creating an unreachable, orphaned history. + if not copy_this_history_id: + raise glx_exceptions.AuthenticationRequired("You need to be logged in to create histories.") + if not trans.galaxy_session: + raise glx_exceptions.AuthenticationRequired("You need an active session to copy histories.") if trans.user and trans.user.bootstrap_admin_user: raise glx_exceptions.RealUserRequiredException("Only real users can create histories.") hist_name = None diff --git a/lib/galaxy_test/api/test_histories.py b/lib/galaxy_test/api/test_histories.py index f605d41ebe0..897efa8fda8 100644 --- a/lib/galaxy_test/api/test_histories.py +++ b/lib/galaxy_test/api/test_histories.py @@ -1,10 +1,14 @@ import time from typing import ClassVar from unittest import SkipTest +from urllib.parse import urljoin from uuid import uuid4 import pytest -from requests import put +from requests import ( + post, + put, +) from galaxy.model.unittest_utils.store_fixtures import ( history_model_store_dict, @@ -638,6 +642,19 @@ class TestHistoriesApi(ApiTestCase, BaseHistories): } self.dataset_populator.import_history(import_data) + def test_anonymous_without_session_cannot_copy_published(self): + history_id = self.dataset_populator.new_history(name=f"for_copying_without_session_{uuid4()}") + self.dataset_populator.make_public(history_id) + + # A request with neither an API key nor a galaxysession cookie has no session to own the copy, + # so copying a published history is rejected rather than doing the copy and orphaning it. + # Regression test for https://github.com/galaxyproject/galaxy/issues/23148. + copy_response = post( + urljoin(self.url, "api/histories"), + json={"history_id": history_id, "name": f"copied_without_session_{uuid4()}"}, + ) + self._assert_status_code_is(copy_response, 403) + def test_publish_non_alphanumeric(self): history_name = "تاریخچه" history_id = self.dataset_populator.new_history(name=history_name) diff --git a/test/unit/app/test_work_context.py b/test/unit/app/test_work_context.py deleted file mode 100644 index 99064f07b85..00000000000 --- a/test/unit/app/test_work_context.py +++ /dev/null @@ -1,39 +0,0 @@ -from unittest import mock - -from galaxy.work.context import SessionRequestContext - - -def _make_context(galaxy_session=None): - return SessionRequestContext( - app=mock.MagicMock(), - request=mock.MagicMock(), - response=mock.MagicMock(), - galaxy_session=galaxy_session, - ) - - -def test_set_history_without_galaxy_session_does_not_raise(): - """Regression test for https://github.com/galaxyproject/galaxy/issues/23148. - - API requests authenticated via API key have no galaxy_session, so set_history - must not unconditionally add/commit a None session to the sa_session. - """ - trans = _make_context(galaxy_session=None) - history = mock.MagicMock(deleted=False) - - trans.set_history(history) - - trans.sa_session.add.assert_not_called() - trans.sa_session.commit.assert_not_called() - - -def test_set_history_with_galaxy_session_updates_and_commits(): - galaxy_session = mock.MagicMock() - trans = _make_context(galaxy_session=galaxy_session) - history = mock.MagicMock(deleted=False) - - trans.set_history(history) - - assert galaxy_session.current_history == history - trans.sa_session.add.assert_called_once_with(galaxy_session) - trans.sa_session.commit.assert_called_once()