TORCH Extraction Implementation

Developer reference for how Aether drives a TORCH data-extraction job end to end: submitting a CRTDL query, polling for completion, and downloading the resulting FHIR NDJSON files.

This page documents the implementation — the types, functions, and control flow — behind Aether's use of TORCH's asynchronous Bulk Data extraction API. For operator-facing usage see the TORCH Integration guide and the TORCH Import step.

TORCH's two APIs

TORCH exposes two REST APIs, and Aether uses only the first:

• FHIR controller — the extraction API, built on the FHIR Asynchronous Bulk Data Request Pattern: the $extract-data kick-off operation plus the GET /fhir/__status/{jobId} status/manifest endpoint.
• Task API (task-controller) — controls extraction jobs in execution (inspect, pause, resume, cancel), exposing each job as a FHIR Task. Aether does not use it today; job control and Task reads are out of scope (see PR #474).

The asynchronous bulk extraction flow

Aether drives the FHIR controller through three phases:

1. Kick-off — POST $extract-data creates the extraction job. TORCH returns the job's status URL (.../fhir/__status/{jobId}) in the Content-Location response header.
2. Poll — Aether polls the status URL. 202 Accepted means the job is still running; 200 OK means it completed and the response body is the async-bulk manifest carrying the output file URLs.
3. Download — Aether downloads each output file.

The extraction job is identified by a TORCH job ID (a UUID, distinct from the Aether pipeline job ID). Aether persists a job handle — the job ID plus its status URL — so it can re-attach to an in-flight job after a crash or restart instead of re-submitting (see Handle persistence and resume below). The CONTEXT.md glossary at the repository root defines the domain terms (extraction job, job handle, re-attach, orphan, re-roll).

Extraction job status (the "double status")

Each extraction job moves through a status lifecycle (coding system https://medizininformatik-initiative.de/torch/job-status):

Status	Meaning
PENDING	Accepted, not yet started
RUNNING_GET_COHORT	Selecting the cohort
RUNNING_PROCESS_BATCH	Extracting patient batches
RUNNING_PROCESS_CORE	Assembling shared (non-patient) core resources
PAUSED	Paused, resumable
TEMP_FAILED	Transient failure — not terminal; TORCH retries
COMPLETED / FAILED / CANCELLED	Terminal

This status appears in two places — an artifact of the async-bulk design that carries job state alongside the Task model: the Task API exposes it as a Task business status, and the completion manifest from __status embeds the same job state in a torch-job extension. Aether reads neither representation. It branches purely on the async-bulk HTTP status the __status endpoint returns (see the Polling stage below): TEMP_FAILED surfaces as 202/503, terminal failure as 500, and completion as 200.

Source map

File	Responsibility
internal/pipeline/import.go	Orchestration: submit → persist handle → poll → download, resume, error classification
internal/pipeline/crtdl_prep.go	PrepareCRTDL — copies/enriches the CRTDL into the job directory before submission
internal/services/torch_client.go	TORCHClient: submit, download, result parsing, file-availability checks, URL resolution, request/response types
internal/services/torch_poller.go	PollConfig and handlePollResponse: status interpretation, liveness window, exponential backoff
internal/models/config.go	TORCHConfig — the tunable knobs
internal/models/job.go	CRTDLPath, TORCHExtractionURL, TORCHJobID — the persisted resume handle
internal/models/step.go	StepTorchImport step constant

Lifecycle

                          internal/pipeline/import.go
┌──────────────────────────────────────────────────────────────────────┐
│ importStep.Run  (step == torch)                                      │
│   InputTypeTORCHURL ──► executeTORCHDownload  (poll URL directly)    │
│   InputTypeCRTDL    ──► executeTORCHExtraction                       │
└───────────────────────────────┬──────────────────────────────────────┘
                                 ▼
        job.TORCHJobID set? ──yes──► re-attach to in-flight job
                 │ no
                 ▼
   submitAndPersistTORCHHandle
     SubmitExtraction ─► POST {base_url}/fhir/$extract-data  (DoOnce, no retry)
        └─ 200 + Content-Location header ─► status URL
     persist job.TORCHExtractionURL + job.TORCHJobID to state.json
                 │
                 ▼
   PollExtractionStatus(statusURL)                 internal/services/torch_poller.go
     loop:
       GET statusURL (DoOnce)
       handlePollResponse:
         202 / 102 ─► in-progress; log OperationOutcome diagnostics; backoff
         200       ─► parseExtractionResult ─► [fileURLs]  ✔ done
         404 / 410 ─► ErrHandleDead ─► clear handle, re-submit, re-poll
         500       ─► terminal job failure (non-transient)
         408/429/other 5xx ─► transient; sleep + backoff; retry
       RecordContact() on every 200/202 (resets liveness window)
       CheckTimeout(): give up only after extraction_timeout of *silence*
                 │
                 ▼
   DownloadExtractionFiles([fileURLs], importDir)  internal/services/torch_client.go
     per file:
       waitForFileAvailability (HEAD, fallback Range GET)
       downloadFile ─► stall-guarded stream ─► {job}/import/<name>.ndjson[.zst]

Stage reference

1. CRTDL preparation

Before the import step runs, PrepareCRTDL (internal/pipeline/crtdl_prep.go) copies the input CRTDL into the job directory as crtdl.json, or — when CRTDL preprocessing is enabled with at least one enrichment — writes an enriched enriched-crtdl.json. It repoints job.CRTDLPath at that file so every downstream step shares one effective CRTDL. Enrichment adds attributes DIMP needs (for example, Patient.identifier) that the original query may omit.

2. Submission

TORCHClient.SubmitExtraction(crtdlPath):

• Reads the CRTDL, validates it is JSON, and base64-encodes it (encodeCRTDLToBase64).
• Wraps it in a FHIR Parameters resource with a single crtdl parameter carrying valueBase64Binary (TORCHExtractionRequest / TORCHParameter).
• POST {base_url}/fhir/$extract-data with Content-Type: application/fhir+json and auth applied via HTTPClient.ApplyAuth.
• Sends with DoOnce — deliberately no retry. $extract-data is non-idempotent job creation; a retried timeout could spawn a duplicate extraction on the server.
• Reads the Content-Location header (the status URL) and normalizes it to an absolute URL (makeAbsoluteURL). A missing header is a hard error.

SubmitExtractionWithContent(crtdlContent []byte) is the in-memory variant for already-enriched documents; it is otherwise identical (it sends Content-Type: application/json).

3. Handle persistence and resume

submitAndPersistTORCHHandle stores the extraction handle before the long poll begins:

• job.TORCHExtractionURL — the status URL to poll.
• job.TORCHJobID — the URL's trailing path segment (JobIDFromStatusURL), for example .../fhir/__status/{jobId}.

Both are written to state.json via UpdateJob, so a crash mid-extraction leaves a recoverable handle. On resume, executeTORCHExtraction sees a non-empty job.TORCHJobID and re-attaches to the running job instead of submitting a new one.

If polling later reports the handle is dead (ErrHandleDead, from a 404/410), the orchestrator clears both fields, re-submits a fresh extraction, and resumes polling.

4. Polling

TORCHClient.PollExtractionStatus(statusURL, showProgress) runs the poll loop; internal/services/torch_poller.go owns the per-response logic and timing state (PollConfig).

Each iteration issues a single GET (DoOnce — the loop, not the HTTP client, owns retry cadence) with Accept: application/json, then handlePollResponse interprets the status:

Status	Meaning	Action
202 Accepted, 102 Processing	Still running	Read OperationOutcome information diagnostics for progress display; back off and poll again
200 OK	Complete	parseExtractionResult → file URLs → return
404 Not Found, 410 Gone	Handle gone	Return ErrHandleDead (caller re-submits)
500 Internal Server Error	Terminal job failure	Non-transient error, stop
408, 429, other 5xx	Transient	Sleep + backoff, retry
other 4xx	Terminal	Stop

Note: 500 is treated as terminal here even though 5xx is normally retryable, because TORCH surfaces a temporary failure (TEMP_FAILED) as 202/503 and reserves 500 for a genuinely failed job.

Liveness window. extraction_timeout is a no-progress window, not a total cap. CheckTimeout returns true only when time.Since(LastContact) > Timeout, and RecordContact resets LastContact on every 200/202. A healthy job that keeps answering — even one running for hours — never trips the timeout; Aether gives up only when TORCH goes silent. See ADR 0001 for the rationale.

Exponential backoff. The poll interval starts at polling_interval and doubles after each in-progress poll (CalculateNextPollInterval / UpdateInterval), capped at max_polling_interval.

Transient network failures on the GET itself (timeouts, connection resets) are swallowed and retried — the extraction may still be running server-side, and the liveness window is the safety net.

5. Result parsing

parseExtractionResult accepts two shapes and extracts absolute file URLs:

• FHIR Parameters (TORCHExtractionResult): reads output parameters and their url parts (extractURLsFromFHIRFormat).
• Async-bulk manifest (TORCHSimpleResponse): { "requiresAccessToken": …, "output": [ { "type": …, "url": … } ], "extension": [ … ] } (extractURLsFromSimpleFormat) — the format the server actually returns today. Aether reads only the output URLs; the manifest's torch-job / diagnostics extensions are ignored.

An empty output is not silently accepted: if TORCH reports errors the parser surfaces them; otherwise it returns a "no matching data" message explaining the likely CRTDL causes (criteria matched nothing, out-of-range time period, unknown cohort).

URL resolution. makeAbsoluteURL is a pure function that dispatches on the URL shape: absolute URLs pass through, path-relative URLs (/output/x.ndjson) resolve against base_url (resolvePathRelativeURL), and scheme-less host-prefixed URLs (host:8080/…, a known TORCH misconfiguration) inherit base_url's scheme (prependBaseScheme).

6. Download

TORCHClient.DownloadExtractionFiles writes each output file into the job's import/ directory. Per file:

• Derive the filename from the URL's base, forcing an .ndjson suffix, then add the compression extension via GetCompressedFilename.
• waitForFileAvailability polls the file before downloading (TORCH results are often served through a proxy with eventual consistency). It uses a HEAD request, falling back to a Range: bytes=0-0 GET when the server answers 403, 404, or 405. It retries up to file_ready_retries times spaced by file_ready_interval; setting file_ready_retries <= 0 disables the check.
• downloadFile / downloadFileOnce stream the body with Accept: application/fhir+ndjson and optional zstd compression on write.

Stall watchdog. Downloads use a dedicated http.Client (downloadClient) with no whole-request deadline, so an arbitrarily large but steadily flowing NDJSON completes regardless of total size. Inactivity is bounded instead by a stallGuardReader: each read re-arms a time.AfterFunc timer, and if no bytes arrive within download_stall_timeout the timer cancels the request context. The cancellation is reported as errDownloadStalled rather than a generic "context canceled". The watchdog is armed before even an error body is read, so a proxy that flushes 4xx/5xx headers then goes silent is bounded too.

Retry. downloadFile retries only a retryable *TORCHError (a transient one), using the shared client's backoff config. A stall or a mid-body write error is not a *TORCHError and is not retried — restarting from scratch would only repeat the stall or waste a full re-transfer.

Each downloaded file yields a models.FHIRDataFile with size, resource (LineCount) count, and SourceStep: StepTorchImport.

7. Error classification

classifyImportError decides whether a failure is transient (pipeline may retry / pause and resume) or terminal:

• It unwraps a *services.TORCHError (via errors.As, so wrapped submit/poll/download errors are caught) and returns its ErrorType.
• For torch/http import steps, a bare network error is transient.
• Everything else defaults to non-transient.

TORCHError carries Operation (submit/poll/download), StatusCode, Message, and ErrorType; IsRetryable() is true only for models.ErrorTypeTransient.

Direct TORCH result URL

When the input is an already-complete TORCH result URL (InputType == InputTypeTORCHURL), executeTORCHDownload skips submission entirely: it polls the URL directly (expecting an immediate 200) and downloads the listed files.

Data types

Defined in internal/services/torch_client.go:

Type	Purpose
TORCHExtractionRequest / TORCHParameter	Kickoff body: FHIR Parameters with a base64 crtdl parameter
TORCHExtractionResult / TORCHResultParameter / TORCHResultPart	FHIR Parameters result format
TORCHSimpleResponse / TORCHSimpleOutput	Simplified { output: [ { type, url } ] } result format
OperationOutcome / OperationOutcomeIssue	Progress diagnostics parsed from 202 bodies
TORCHError	Structured submit/poll/download error with retryability

Sentinel errors: ErrExtractionTimeout, ErrHandleDead, ErrInvalidCRTDL.

Configuration

TORCHConfig (internal/models/config.go) tunes the flow. Defaults are set in DefaultConfig and range-checked in TORCHConfig.Validate:

Field	Default	Effect
base_url	—	TORCH server base URL (required when the torch step is enabled)
username / password	—	Basic auth (takes precedence over OAuth)
oauth_issuer_uri / oauth_client_id / oauth_client_secret	—	OAuth 2.0 client-credentials auth
extraction_timeout	30m	Liveness window — max time to wait without a response; reset on every 200/202
polling_interval	5s	Initial status poll interval (must be ≥ 1s)
max_polling_interval	30s	Backoff cap (must be ≥ polling_interval)
file_ready_retries	10	Availability checks before download; 0 disables the check
file_ready_interval	10s	Delay between availability checks
download_stall_timeout	60s	Inactivity window while streaming a file; 0 uses the built-in default

See the Configuration Reference for the operator-facing table.

Persisted job handle

For resume, the following PipelineJob fields (internal/models/job.go) are written to state.json:

Field	Purpose
CRTDLPath	The effective CRTDL used for submission (decoupled from the raw input source)
TORCHExtractionURL	The Content-Location status URL to re-poll
TORCHJobID	The extraction job ID (handle) used to re-attach to an in-flight job

Testing seams

The import step depends on the Extractor interface — SubmitExtraction, PollExtractionStatus, DownloadExtractionFiles — which TORCHClient satisfies. extractorFactory (internal/pipeline/import.go) builds it, and SetExtractorFactoryForTesting / ResetExtractorFactory swap in a fake so tests exercise submit/poll/download without a live TORCH server. Pure helpers such as makeAbsoluteURL, CalculateNextPollInterval, and the PollConfig liveness methods are unit-tested directly.

Next Steps

• Architecture — system-wide design and data flow
• TORCH Integration — operator-facing guide
• ADR 0001: Extraction timeout liveness
• Configuration Reference — TORCH config knobs