Implement resumable staging of foreign files from HTTP sources

[scwatts]

Summary

Hello, I work on the nf-core/oncoanalyser pipeline and I'm opening this PR as an extension of the (great!) work done in response to #5214

The motivation for this small PR is that during normal operation of oncoanalyser, the pipeline may retrieve a large amount of data (sometimes hundreds of GBs) from Cloudflare R2 via HTTP. This service works well for us and our users with respect to their egress-free model, however as mentioned in the linked PR, connections to R2 during download of the largest files are often interrupted

The Nextflow file staging retry functionality for HTTP does handle this in a reasonable way by deleting the partially retrieved file and then starting the download again, but often is the case for us that each successive download attempt experiences the same connection interruption, leading to complete failure of the Nextflow / oncoanalyser run. Ideally, the retry would resume the partially downloaded file so that progress is made during each retry

There was some consideration for such a feature here, but it (fairly) hadn't been picked up since

In this PR I implement an example of the retry via resume functionality for HTTP connections, and extend the mechanism to also include S3 multipart upload destinations. I would be grateful if you were to consider merging, or otherwise implementing this resume-retry feature. It would make a real difference for UX in oncoanalyser and for the use of Cloudflare R2 in general

Disclaimer: I have used Claude Code to help create this PR, and while each line of code has been reviewed and the design carefully considered, I am still largely naive to the Nextflow codebase

Details of Pull Request

Click to show

Resume HTTP staging downloads with a byte range

When Nextflow stages a foreign file from an HTTP(S) URL and the transfer fails partway through, the retry discards the partially-downloaded file and starts over from byte 0. For large files on unstable connections this throws away every byte already transferred when the connection is interrupted

These changes add a mechanism where on retry after connection interruption, an attempt is made to resume from the last downloaded byte using an HTTP Range header. If the server ignores the range (returning 200 instead of 206), it falls back to re-downloading from the beginning matching existing behaviour. When a file is resume-retried, it still consumes a maxRetries count. Resume applies to HTTP/HTTPS only; FTP and other foreign sources keep the existing delete-and-restart behaviour

Resume S3 multipart upload destinations

When the staging target is an S3 object, the same interruption leaves an in-progress multipart upload rather than a local partial file. The S3 provider now exposes that upload as a ResumableUpload, and the HTTP source resumes the upload from the first uncommitted part instead of aborting the multipart upload and restarting from the first byte

Changes

Click to show

Top-to-bottom along the staging call path:

• nextflow FilePorter: the retry loop now retries on recoverable IOExceptions (fail-fast on a missing source and on interruption) and detects a resumable partial file (HTTP(S) source + partial target), passing the new HttpCopyOption.RESUME flag down; otherwise it deletes the partial file and restarts. A resume consumes a maxRetries count, and a terminal failure aborts any in-progress resumable upload
• nf-commons: new ResumableFileSystem and ResumableUpload interfaces plus the HttpCopyOption.RESUME flag. FileHelper.copyPath is unchanged; RESUME flows through its existing FileSystemTransferAware dispatch to download()
• nf-httpfs XFileSystemProvider.download: a plain copy honours copy options (replace, or throw if the target exists). A RESUME copy sends a Range header, validates the Content-Range, appends on 206, and falls back to a full re-download on 200/416/mismatch; for a resumable target it resumes the in-progress upload instead of appending to a local file
• nf-amazon S3FileSystemProvider: implements ResumableFileSystem. newUpload starts an S3 multipart upload; resumeUpload lists and recovers the in-progress multipart upload (with pagination) so a retry continues from the last committed byte instead of re-uploading from byte 0
• nf-amazon S3OutputStream: resumes an existing multipart upload from a set of recovered parts, and abandon() leaves the upload in-progress (flushing only a full minimum part) so a later attempt can resume it

Unit tests cover resume/restart and Content-Range/416 edge cases (XFileSystemProviderTest), the resume decision (FilePorterTest), the S3 multipart resume and error propagation (S3FileSystemProviderTest), and the resumable stream (S3OutputStreamTest). ./gradlew :nf-httpfs:test :nextflow:test :plugins:nf-amazon:test passes

Testing

Click to show

See below Files section for scripts referenced in below testing commands

A Python server simulates the mid-transfer failure: it advertises the full Content-Length but drops the connection mid-body, leaving a partial file. For files larger than --part-size (default 10 MiB, matching the fixed 10 MiB buffer of Nextflow's S3OutputStream) it truncates after one full multipart part plus half of the next, so an S3 multipart upload has at least one committed part to resume

The new code in this PR then makes a ranged request to which the server honours and returns the remaining bytes (or the full body with --ignore-range, to exercise the restart fallback)

Given requirements around size of served file, I generate a ~50 MiB payload for testing, noting its MD5 hash:

mkdir -p data/

yes 'nextflow-io/nextflow: A DSL for data-driven computational pipelines' | \
  awk '{ print NR, $0 }' | \
  awk -v bytes=$((50 * 1024 * 1024)) '{ n += length($0)+1; print; if (n >= bytes) exit }' > data/input.txt

md5sum data/input.txt | cut -f1 -d' '
#5b169a50e13cee1f0604e9edd9c7747

Standard resume (HTTP server honours range request)

Click to show

Start the server:

./scripts/file_server.py --file data/input.txt --port 8000

Run:

./nextflow-src/launch.sh run scripts/main.nf -ansi-log false --url http://localhost:8000/input.txt --md5 5b169a50e13cee1f0604e9edd9c7747b

The logs shows a byte-range resume rather than a byte-0 restart:

Nextflow:
WARN: Unable to stage foreign file: http://localhost:8000/input.txt (resuming, attempt 1 of 3) -- Cause: Premature EOF
[c6/ac8f64] Submitted process > STAGE_AND_VERIFY (1)
Input md5:    5b169a50e13cee1f0604e9edd9c7747b
Computed md5: 5b169a50e13cee1f0604e9edd9c7747b
Result: MATCH

Server:
serving input.txt (52428870 bytes, part_size=10485760, prefix=15728640, drop_count=1) on http://localhost:8000 (ignore_range=False)
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [size check]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [download (truncated)]  [chunk size: 30.0%] [total size: 30.0%]
server: "GET /input.txt HTTP/1.1" 206 -  [Range: bytes=15728640-]  [resume (complete)]  [chunk size: 70.0%] [total size: 100.0%]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [size check]

Restart retry (HTTP server ignores range request)

Click to show

Start the server:

./scripts/file_server.py --file data/input.txt --port 8000 --ignore-range

Run:

./nextflow-src/launch.sh run scripts/main.nf -ansi-log false --url http://localhost:8000/input.txt --md5 5b169a50e13cee1f0604e9edd9c7747b

The logs shows a successful byte-0 restart rather than a resume:

Nextflow:
WARN: Unable to stage foreign file: http://localhost:8000/input.txt (resuming, attempt 1 of 3) -- Cause: Premature EOF
[10/99ea7a] Submitted process > STAGE_AND_VERIFY (1)
Input md5:    5b169a50e13cee1f0604e9edd9c7747b
Computed md5: 5b169a50e13cee1f0604e9edd9c7747b
Result: MATCH

Server:
serving input.txt (52428870 bytes, part_size=10485760, prefix=15728640, drop_count=1) on http://localhost:8000 (ignore_range=True)
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [size check]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [download (truncated)]  [chunk size: 30.0%] [total size: 30.0%]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: bytes=15728640-]  [resume (rejected)]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [download (full)]  [chunk size: 100.0%] [total size: 100.0%]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [size check]

AWS S3 destination

Click to show

To exercise the S3 multipart resume, serve a payload of at least two multipart parts (≥ 20 MiB) and stage it into an S3 work dir with -w s3://bucket-name/work

Other approaches for local testing such as MinIO could be used here, but I decided to test directly on AWS S3

First create a Docker image with typical coreutils, libz (aws dep), and AWS CLIv2:

docker build \
  --platform linux/amd64 \
  --push \
  --file scripts/Dockerfile \
  --tag docker.io/scwatts/awscliv2:20260817--0 \
  .

Then place AWS credentials into BASH environment:

export AWS_ACCESS_KEY_ID=''
export AWS_SECRET_ACCESS_KEY=''
export AWS_DEFAULT_REGION=''

Re-launch the file server to reset drop count:

./scripts/file_server.py --file data/input.txt --port 8000

Then run Nextflow script, supplying the AWS Batch conifguration:

./nextflow-src/launch.sh run scripts/main.nf -ansi-log false -config scripts/aws_batch.config --url http://localhost:8000/input.txt --md5 5b169a50e13cee1f0604e9edd9c7747b

Log shows good resume in multipart upload context:

Nextflow:
Staging foreign file: http://localhost:8000/input.txt
WARN: Unable to stage foreign file: http://localhost:8000/input.txt (resuming, attempt 1 of 3) -- Cause: Premature EOF
[12/adb37f] Submitted process > STAGE_AND_VERIFY (1)
Input md5:    5b169a50e13cee1f0604e9edd9c7747b
Computed md5: 5b169a50e13cee1f0604e9edd9c7747b
Result: MATCH

Server:
serving input.txt (52428870 bytes, part_size=10485760, prefix=15728640, drop_count=1) on http://localhost:8000 (ignore_range=False)
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [size check]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [download (truncated)]  [chunk size: 30.0%] [total size: 30.0%]
server: "GET /input.txt HTTP/1.1" 206 -  [Range: bytes=15728640-]  [resume (complete)]  [chunk size: 70.0%] [total size: 100.0%]
server: "GET /input.txt HTTP/1.1" 200 -  [Range: <none>]  [size check]

Files

File: scripts/file_server.py (click to show)

#!/usr/bin/env python3
import argparse
import http.server
import pathlib
import sys
import threading


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--file', required=True, type=pathlib.Path, help='payload to serve')
    parser.add_argument('--port', type=int, default=8000, help='TCP port to listen on')
    parser.add_argument('--part-size', type=int, default=10485760,
                        help='multipart part size in bytes (matches the fixed 10 MiB S3OutputStream buffer in Nextflow)')
    parser.add_argument('--ignore-range', action='store_true', help='Always return 200 on Range requests')
    parser.add_argument('--drop-count', type=int, default=1,
                        help='number of download attempts to interrupt before serving a complete response')

    args = parser.parse_args()
    if not args.file.exists():
        parser.error(f'Input file {args.file} does not exist')

    return args


def load_payload(path):
    with path.open('rb') as f:
        return f.read()


# The first (non-ranged) response is truncated at prefix_len bytes. For the truncation to be
# reliably distinguishable from a header-only read (which disconnects after reading the status and
# headers), prefix_len must exceed the OS socket send/receive buffers, so writing the truncated body
# raises BrokenPipeError when the client has already gone. 1 MiB comfortably exceeds the default
# buffers (128-256 KiB on localhost, up to a few MiB when auto-tuned). The prefix is at least half
# the file (and at least part_size for large files), so a file of 2 MiB or more is always safe.
MIN_SAFE_PREFIX = 1 * 1024 * 1024


def serve(file_name, data, port, ignore_range, part_size, drop_count):

    size = len(data)
    # Truncate inside the second multipart part, so one full part is already committed and can be
    # resumed from; fall back to half the file when it is too small to produce a multipart upload
    if size > part_size:
        prefix_len = min(size - 1, part_size + part_size // 2)
    else:
        prefix_len = max(1, size // 2)

    # Drop count remaining. Nextflow issues header-only GETs (readAttributes) that read just the status
    # and headers before the real download, so a drop is only counted when its truncated body is
    # actually delivered to a client that read it; a client that hung up mid-body is refunded.
    drops_remaining = [drop_count]
    drops_lock = threading.Lock()
    total_sent = [0]  # cumulative bytes delivered across requests

    class Handler(http.server.BaseHTTPRequestHandler):

        def _write(self, body):
            # Return False when the client disconnects mid-body (a header-only read or an abort),
            # which is expected for a truncating test server and not an error
            try:
                self.wfile.write(body)
                self.wfile.flush()
                total_sent[0] += len(body)
                return True
            except (BrokenPipeError, ConnectionResetError):
                return False


        def _send(self, code, body, extra=(), range_ignored=False):
            self._status = code
            self._payload = len(body)
            self._range_ignored = range_ignored
            self._label = 'resume (complete)' if code == 206 else 'download (full)'
            self.send_response(code)
            for k, v in extra:
                self.send_header(k, v)
            self.send_header('Content-Length', str(len(body)))
            self.end_headers()
            if self._write(body):
                if code == 200:  # fresh byte-0 download starts a new run
                    total_sent[0] = len(body)
            else:
                self._payload = 0
                self._label = 'size check'


        def _drop(self, range_header):
            # Truncate the body mid-stream and close; report whether the body was delivered
            # ignore_range means the server never honours Range, so a ranged request is treated
            # as a full byte-0 download here, exactly like a non-ranged one
            self._range_ignored = bool(range_header) and ignore_range
            if not range_header or ignore_range:
                self._status = 200
                self._payload = len(data[:prefix_len])
                self._label = 'download (truncated)'
                self.send_response(200)
                self.send_header('Content-Length', str(size))  # advertise the full size
                self.end_headers()
                delivered = self._write(data[:prefix_len])
                if delivered:
                    total_sent[0] = len(data[:prefix_len])  # fresh byte-0 download starts a new run
                else:
                    self._payload = 0
                    self._label = 'size check'
                self.connection.close()
                return delivered

            start = int(range_header.split('=')[1].split('-')[0])
            remaining = size - start
            body = data[start:start + remaining // 2]
            self._status = 206
            self._payload = len(body)
            self._label = 'resume (truncated)'
            self.send_response(206)
            self.send_header('Content-Range', f'bytes {start}-{size-1}/{size}')
            self.send_header('Content-Length', str(remaining))
            self.end_headers()
            delivered = self._write(body)
            if not delivered:
                self._payload = 0
                self._label = 'size check'
            self.connection.close()
            return delivered


        def do_GET(self):
            range_header = self.headers.get('Range')

            with drops_lock:
                drop = drops_remaining[0] > 0
            if drop:
                if self._drop(range_header):
                    with drops_lock:
                        drops_remaining[0] -= 1
            elif not range_header:
                self._send(200, data)
            else:
                start = int(range_header.split('=')[1].split('-')[0])
                if not ignore_range:
                    self._send(206, data[start:], (('Content-Range', f'bytes {start}-{size-1}/{size}'),))
                else:
                    self._send(200, data, range_ignored=True)

            self._log(range_header)

        def _log(self, range_header):
            payload = getattr(self, '_payload', 0)
            status = getattr(self, '_status', '-')
            label = getattr(self, '_label', '?')
            if payload == 0:
                # Header-only read: the client read the status and headers then disconnected.
                # readHttpAttributes (no Range) is a size check; a Range request answered 200
                # instead of 206 is resumeDownload rejecting the resume and falling back to restart
                label = 'resume (rejected)' if range_header else 'size check'
            elif getattr(self, '_range_ignored', False):
                label += ', range ignored'
            rng = f'  [Range: {range_header}]' if range_header else '  [Range: <none>]'
            if payload == 0:
                sys.stderr.write(f'server: "{self.requestline}" {status} -{rng}  [{label}]\n')
            else:
                pct = (payload / size * 100) if size else 0.0
                total_pct = (total_sent[0] / size * 100) if size else 0.0
                sys.stderr.write(f'server: "{self.requestline}" {status} -{rng}  [{label}]  [chunk size: {pct:.1f}%] [total size: {total_pct:.1f}%]\n')

        def log_message(self, fmt, *args):
            # Disable auto-log; _log() runs after the body so the cumulative total is accurate
            pass

    print(f'serving {file_name} ({size} bytes, part_size={part_size}, prefix={prefix_len}, drop_count={drop_count}) '
          f'on http://localhost:{port} (ignore_range={ignore_range})')
    http.server.ThreadingHTTPServer(('0.0.0.0', port), Handler).serve_forever()


def main():
    # Get commandline arguments and input file to serve
    args = parse_args()
    data = load_payload(args.file)

    # Require minimum size, needed for initial buffer fill
    if len(data) < 2 * MIN_SAFE_PREFIX:
        sys.exit(
            f'error: {args.file} is too small ({len(data)} bytes); need at least '
            f'{2 * MIN_SAFE_PREFIX} bytes so the truncated prefix exceeds the OS socket buffer'
        )

    # Launch file server
    serve(args.file.name, data, args.port, args.ignore_range, args.part_size, args.drop_count)


if __name__ == '__main__':
    main()

File: scripts/main.nf (click to show)

params.url = null
params.md5 = null

process STAGE_AND_VERIFY {
    input:
    path fp

    output:
    stdout emit: hash

    script:
    """
    md5sum ${fp} | cut -d' ' -f1
    """
}

workflow {
  main:

  STAGE_AND_VERIFY(
    channel.fromPath(params.url),
  )

  STAGE_AND_VERIFY.out.hash
    .map { hash_stdout ->
      def hash = hash_stdout.trim()
      def result = hash == params.md5 ? 'MATCH' : 'MISMATCH'
      println "Input md5:    ${params.md5}"
      println "Computed md5: ${hash}"
      println "Result: ${result}"
    }
}

File: scripts/Dockerfile (click to show)

FROM continuumio/miniconda3:26.5.3 AS build

RUN \
    echo > ~/.condarc '\
channels:\n\
    - conda-forge\n\
    - bioconda\n\
    - defaults'

RUN \
    conda create -y -p /build/ curl unzip

RUN \
    conda create -y -p /env/ libzlib python

RUN \
    conda clean -yaf

# Move Conda environment into standard BioContainers base image
# Then install AWS CLIv2; here in this final container in order to link against correct libraries
FROM quay.io/bioconda/base-glibc-busybox-bash:3.1

COPY --from=build /env/ /env/
COPY --from=build /build/ /build/

RUN \
  mkdir -p /tmp/awscliv2/ && cd /tmp/awscliv2/ && \
    /build/bin/curl https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip -o awscliv2.zip && \
    /build/bin/unzip awscliv2.zip && \
    ./aws/install

ENV PATH="/env/bin:${PATH}"
ENV LD_LIBRARY_PATH="/env/lib/"

RUN \
  rm -r /build/ /tmp/awscliv2/

File: scripts/aws_batch.config (click to show)

process {
  executor = 'awsbatch'
  queue = 'batch-queue-name'
  container = 'docker.io/scwatts/awscliv2:20260817--0'
}

aws {
  region = 'aws-region-name'
}

workDir = 's3://bucket-name/object-prefix/'

[netlify]

✅ Deploy Preview for nextflow-docs ready!

Name	Link
🔨 Latest commit	53fcea3
🔍 Latest deploy log	https://app.netlify.com/projects/nextflow-docs/deploys/6a82a8c2b95af80008fb5818
😎 Deploy Preview	https://deploy-preview-7494--nextflow-docs.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.