From fae0766d40343554458f9ae3492f4e489ceb333e Mon Sep 17 00:00:00 2001 From: Stephen Watts Date: Thu, 13 Aug 2026 15:30:24 +1000 Subject: [PATCH 1/2] Implement resume for HTTP file staging retry Assisted-by: Claude Code Signed-off-by: Stephen Watts --- .../groovy/nextflow/file/FilePorter.groovy | 48 +++- .../nextflow/file/FilePorterTest.groovy | 221 ++++++++++++++++ .../main/nextflow/file/HttpCopyOption.groovy | 28 ++ .../file/http/XFileSystemProvider.groovy | 139 +++++++++- .../file/http/XFileSystemProviderTest.groovy | 242 ++++++++++++++++++ 5 files changed, 659 insertions(+), 19 deletions(-) create mode 100644 modules/nf-commons/src/main/nextflow/file/HttpCopyOption.groovy diff --git a/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy b/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy index 0f278a7ba8..dd628ac53e 100644 --- a/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy @@ -16,6 +16,7 @@ package nextflow.file +import java.nio.file.FileSystems import java.nio.file.FileVisitResult import java.nio.file.Files import java.nio.file.NoSuchFileException @@ -323,16 +324,28 @@ class FilePorter { protected Path stageForeignFile(Path filePath, Path stagePath) { int count = 0 + boolean resume = false while( true ) { try { - return stageForeignFile0(filePath, stagePath) + // only the first attempt may short-circuit on an existing (cached) target; + // retries must force a download so that a partial file can be resumed + return count == 0 + ? stageForeignFile0(filePath, stagePath) + : copyForeignFile(filePath, stagePath, resume) } catch( IOException e ) { - // remove the target file that could be have partially downloaded - cleanup(stagePath) // check if a stage/download retry is allowed - if( count++ < maxRetries && recoverableError(e) && !Thread.currentThread().isInterrupted() ) { - def message = "Unable to stage foreign file: ${filePath.toUriString()} (try ${count} of ${maxRetries}) -- Cause: $e.message" + final retry = count < maxRetries && !Thread.currentThread().isInterrupted() + // resume a partial download (HTTP/HTTPS only), otherwise discard the + // partially downloaded file and start again from the beginning + resume = retry && canResumeDownload(filePath, stagePath) + if( !resume ) + cleanup(stagePath) + if( retry ) { + count++ + def message = resume + ? "Unable to stage foreign file: ${filePath.toUriString()} (resuming, attempt ${count} of ${maxRetries}) -- Cause: $e.message" + : "Unable to stage foreign file: ${filePath.toUriString()} (try ${count} of ${maxRetries}) -- Cause: $e.message" log.isDebugEnabled() ? log.warn(message, e) : log.warn(message) sleep (10 + RND.nextInt(300)) @@ -344,13 +357,15 @@ class FilePorter { } } - private boolean recoverableError(IOException e){ - final result = - e !instanceof NoSuchFileException - && (e instanceof SocketTimeoutException || e !instanceof InterruptedIOException) - && e !instanceof SocketException - log.debug "Stage foreign file exception: recoverable=$result; type=${e.class.name}; message=${e.message}" - return result + @PackageScope + boolean canResumeDownload(Path source, Path target) { + // a resume only makes sense for a partially downloaded HTTP(S) file written to a local + // target, where APPEND is supported + if( !Files.exists(target) || Files.size(target) == 0 ) + return false + if( target.fileSystem != FileSystems.getDefault() ) + return false + return source.toUri().scheme in ['http', 'https'] } private String fmtError(Path filePath, Exception e) { @@ -367,10 +382,17 @@ class FilePorter { log.debug "Local cache found for foreign file ${source.toUriString()} at ${target.toUriString()}" return target } + return copyForeignFile(source, target, false) + } + + @PackageScope + Path copyForeignFile(Path source, Path target, boolean resume) { log.debug "Copying foreign file ${source.toUriString()} to work dir: ${target.toUriString()}" if( debugDelay ) sleep ( new Random().nextInt(debugDelay) ) - return FileHelper.copyPath(source, target) + return resume + ? FileHelper.copyPath(source, target, HttpCopyOption.RESUME) + : FileHelper.copyPath(source, target) } synchronized String getMessageAndClear() { diff --git a/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy b/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy index 66a7798381..6917d16d05 100644 --- a/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy @@ -21,12 +21,16 @@ import java.nio.file.Files import java.nio.file.Path import java.util.concurrent.Semaphore +import com.github.tomakehurst.wiremock.junit.WireMockRule import groovy.util.logging.Slf4j import nextflow.Session import nextflow.exception.ProcessStageException +import org.junit.Rule import spock.lang.Ignore import spock.lang.Specification import test.TestHelper + +import static com.github.tomakehurst.wiremock.client.WireMock.* /** * * @author Paolo Di Tommaso @@ -34,6 +38,9 @@ import test.TestHelper @Slf4j class FilePorterTest extends Specification { + @Rule + WireMockRule wireMockRule = new WireMockRule(0) + def 'should get the max retries value' () { @@ -52,6 +59,220 @@ class FilePorterTest extends Specification { } + def 'should determine whether a download can resume' () { + + given: + def httpSource = 'http://localhost:1234/file.txt' as Path + def ftpSource = 'ftp://localhost/file.txt' as Path + def localSource = TestHelper.createInMemTempFile('local.txt', 'hello') + + and: + def folder = Files.createTempDirectory('test') + def partial = folder.resolve('partial.txt'); partial.text = 'partial' + def empty = folder.resolve('empty.txt'); empty.text = '' + def missing = folder.resolve('missing.txt') + + and: + def porter = new FilePorter.FileTransfer(httpSource, partial, 3, Mock(Semaphore)) + + expect: + // http source with a partial file -> resumable + porter.canResumeDownload(httpSource, partial) + // no partial file yet -> not resumable + !porter.canResumeDownload(httpSource, missing) + // empty target -> not resumable + !porter.canResumeDownload(httpSource, empty) + // ftp source -> not resumable (no byte-range resume) + !porter.canResumeDownload(ftpSource, partial) + // local source -> not resumable + !porter.canResumeDownload(localSource, partial) + + cleanup: + folder?.deleteDir() + } + + + def 'should keep a partial file when resuming a retry' () { + + given: + def source = 'http://localhost:1234/file.txt' as Path + def folder = Files.createTempDirectory('test') + def target = folder.resolve('file.txt') + + and: + def transfer = new RetryTransfer(source, target, 3) + + when: + transfer.stageForeignFile(source, target) + + then: + !transfer.cleaned + target.text == 'complete' + + cleanup: + folder?.deleteDir() + } + + + def 'should discard a partial file when retries are exhausted' () { + + given: + def source = 'http://localhost:1234/file.txt' as Path + def folder = Files.createTempDirectory('test') + def target = folder.resolve('file.txt') + + and: + def transfer = new AlwaysFailTransfer(source, target, 3) + + when: + transfer.stageForeignFile(source, target) + + then: + thrown(ProcessStageException) + transfer.cleaned + + cleanup: + folder?.deleteDir() + } + + + def 'should retry when a socket exception interrupts the download' () { + + given: + def source = 'http://localhost:1234/file.txt' as Path + def folder = Files.createTempDirectory('test') + def target = folder.resolve('file.txt') + + and: + def transfer = new SocketResetTransfer(source, target, 3) + + when: + transfer.stageForeignFile(source, target) + + then: + !transfer.cleaned + target.text == 'complete' + + cleanup: + folder?.deleteDir() + } + + + def 'should resume a partial download via the copyPath seam' () { + + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' // 20 bytes + def REMAINING = 'KLMNOPQRST' // remaining 10 bytes + + // first (non-ranged) request returns a truncated body with a full Content-Length + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody('ABCDEFGHIJ'))) // only 10 bytes -> truncated download + + // ranged request returns the remaining bytes + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .atPriority(1) + .withHeader('Range', equalTo('bytes=10-')) + .willReturn(aResponse() + .withStatus(206) + .withHeader('Content-Range', 'bytes 10-19/20') + .withHeader('Content-Length', '10') + .withBody(REMAINING))) + + def source = "${localhost}/file.txt" as Path + def folder = Files.createTempDirectory('test') + def target = folder.resolve('file.txt') + + and: + def transfer = new FilePorter.FileTransfer(source, target, 3, new Semaphore(100)) + + when: + transfer.stageForeignFile(source, target) + + then: + target.text == FULL + + cleanup: + folder?.deleteDir() + } + + + static class RetryTransfer extends FilePorter.FileTransfer { + int attempts = 0 + boolean cleaned = false + + RetryTransfer(Path source, Path target, int maxRetries) { + super(source, target, maxRetries, new Semaphore(100)) + } + + @Override + protected void cleanup(Path path) { + cleaned = true + super.cleanup(path) + } + + @Override + Path copyForeignFile(Path source, Path target, boolean resume) { + if( attempts++ == 0 ) { + target.text = 'partial' + throw new IOException('boom') + } + target.text = 'complete' + return target + } + } + + + static class AlwaysFailTransfer extends FilePorter.FileTransfer { + boolean cleaned = false + + AlwaysFailTransfer(Path source, Path target, int maxRetries) { + super(source, target, maxRetries, new Semaphore(100)) + } + + @Override + protected void cleanup(Path path) { + cleaned = true + super.cleanup(path) + } + + @Override + Path copyForeignFile(Path source, Path target, boolean resume) { + target.text = 'partial' + throw new IOException('boom') + } + } + + + static class SocketResetTransfer extends FilePorter.FileTransfer { + int attempts = 0 + boolean cleaned = false + + SocketResetTransfer(Path source, Path target, int maxRetries) { + super(source, target, maxRetries, new Semaphore(100)) + } + + @Override + protected void cleanup(Path path) { + cleaned = true + super.cleanup(path) + } + + @Override + Path copyForeignFile(Path source, Path target, boolean resume) { + if( attempts++ == 0 ) { + target.text = 'partial' + throw new SocketException('Connection reset') + } + target.text = 'complete' + return target + } + } + + def 'should copy foreign files' () { given: diff --git a/modules/nf-commons/src/main/nextflow/file/HttpCopyOption.groovy b/modules/nf-commons/src/main/nextflow/file/HttpCopyOption.groovy new file mode 100644 index 0000000000..4fedf35167 --- /dev/null +++ b/modules/nf-commons/src/main/nextflow/file/HttpCopyOption.groovy @@ -0,0 +1,28 @@ +/* + * Copyright 2013-2026, Seqera Labs + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package nextflow.file + +import java.nio.file.CopyOption + +/** + * Copy option requesting a byte-range resume of a partially downloaded target. + */ +enum HttpCopyOption implements CopyOption { + + RESUME + +} diff --git a/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy b/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy index 4f084fd479..f5a8558873 100644 --- a/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy +++ b/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy @@ -17,6 +17,9 @@ package nextflow.file.http import nextflow.file.CopyMoveHelper +import nextflow.file.CopyOptions +import nextflow.file.FileSystemTransferAware +import nextflow.file.HttpCopyOption import static nextflow.file.http.XFileSystemConfig.* @@ -26,6 +29,7 @@ import java.nio.file.AccessDeniedException import java.nio.file.AccessMode import java.nio.file.CopyOption import java.nio.file.DirectoryStream +import java.nio.file.FileAlreadyExistsException import java.nio.file.FileStore import java.nio.file.FileSystem import java.nio.file.FileSystemNotFoundException @@ -60,7 +64,7 @@ import sun.net.www.protocol.ftp.FtpURLConnection @Slf4j @PackageScope @CompileStatic -abstract class XFileSystemProvider extends FileSystemProvider { +abstract class XFileSystemProvider extends FileSystemProvider implements FileSystemTransferAware { private Map fileSystemMap = new LinkedHashMap<>(20) @@ -178,18 +182,24 @@ abstract class XFileSystemProvider extends FileSystemProvider { } protected URLConnection toConnection(Path path) { + toConnection(path, 0) + } + + protected URLConnection toConnection(Path path, long offset) { final url = path.toUri().toURL() log.trace "File remote URL: $url" - return toConnection0(url, 0) + return toConnection0(url, 0, offset) } - protected URLConnection toConnection0(URL url, int attempt) { + protected URLConnection toConnection0(URL url, int attempt, long offset) { final conn = url.openConnection() conn.setRequestProperty("User-Agent", 'Nextflow/httpfs') if( conn instanceof HttpURLConnection ) { // by default HttpURLConnection does redirect only within the same host // disable the built-in to implement custom redirection logic (see below) conn.setInstanceFollowRedirects(false) + if( offset > 0 ) + conn.setRequestProperty("Range", "bytes=$offset-") } if( url.userInfo ) { conn.setRequestProperty("Authorization", auth(url.userInfo)); @@ -204,17 +214,17 @@ abstract class XFileSystemProvider extends FileSystemProvider { final newUrl = new URI(absLocation(location,url)).toURL() if( url.protocol=='https' && newUrl.protocol=='http' ) throw new IOException("Refuse to follow redirection from HTTPS to HTTP (unsafe) URL - origin: $url - target: $newUrl") - return toConnection0(newUrl, attempt+1) + return toConnection0(newUrl, attempt+1, offset) } else if( conn instanceof HttpURLConnection && conn.getResponseCode() in config().retryCodes() && attempt < config().maxAttempts() ) { final delay = (Math.pow(config().backOffBase(), attempt) as long) * config().backOffDelay() log.debug "Got HTTP error=${conn.getResponseCode()} waiting for ${delay}ms (attempt=${attempt+1})" Thread.sleep(delay) - return toConnection0(url, attempt+1) + return toConnection0(url, attempt+1, offset) } else if( conn instanceof HttpURLConnection && conn.getResponseCode()==401 && attempt==0 ) { if( XAuthRegistry.instance.refreshToken(conn) ) { - return toConnection0(url, attempt+1) + return toConnection0(url, attempt+1, offset) } } return conn @@ -423,6 +433,123 @@ abstract class XFileSystemProvider extends FileSystemProvider { throw new UnsupportedOperationException("Move not supported by ${getScheme().toUpperCase()} file system provider") } + @Override + boolean canDownload(Path source, Path target) { + // byte-range resume is only supported for HTTP(S); FTP keeps the default copy path + return getScheme() in ['http','https'] + } + + @Override + boolean canUpload(Path source, Path target) { + return false + } + + @Override + void download(Path source, Path target, CopyOption... options) throws IOException { + if( source.class != XPath ) + throw new ProviderMismatchException() + + if( options.contains(HttpCopyOption.RESUME) ) { + // resume a partially downloaded file, falling back to a full download + final long offset = Files.exists(target) ? Files.size(target) : 0 + if( offset > 0 && resumeDownload(source, target, offset) ) + return + downloadFromStart(source, target) + } + else { + // plain copy honouring the standard copy options; the target is only overwritten inside + // downloadFromStart after the source has responded successfully + final CopyOptions opts = CopyOptions.parse(options) + if( Files.exists(target) && !opts.replaceExisting() ) + throw new FileAlreadyExistsException(target.toString()) + downloadFromStart(source, target) + } + } + + /** + * Append the remaining bytes of {@code source} to the partial file at {@code target}. + * + * @return {@code true} when the server honoured the exact requested range and the remaining + * bytes were appended, {@code false} when the server ignored or rejected the range + * and a full re-download is required + */ + private boolean resumeDownload(Path source, Path target, long offset) throws IOException { + final conn = toConnection(source, offset) + if( conn !instanceof HttpURLConnection ) + return false + try { + final int code = ((HttpURLConnection)conn).getResponseCode() + if( code != 206 ) + return false + // only accept a resume when the server honoured the exact requested range through EOF + final long[] range = parseContentRange(conn) + if( range == null || range[0] != offset || range[1] + 1 != range[2] ) + return false + final long remaining = range[2] - offset + try( InputStream in = checkedInputStream(conn, remaining) ) { + try( OutputStream out = Files.newOutputStream(target, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.APPEND) ) { + copyStream(in, out) + } + } + return true + } + finally { + ((HttpURLConnection)conn).disconnect() + } + } + + private void downloadFromStart(Path source, Path target) throws IOException { + final conn = toConnection(source) + if( conn !instanceof HttpURLConnection ) + throw new IOException("Download not supported for non-HTTP source: ${FilesEx.toUriString(source)}") + try { + final int code = ((HttpURLConnection)conn).getResponseCode() + if( code != 200 ) + throw new IOException("Unable to download foreign file ${FilesEx.toUriString(source)} -- unexpected HTTP status code: $code") + try( InputStream in = checkedInputStream(conn, conn.getContentLengthLong()) ) { + try( OutputStream out = Files.newOutputStream(target, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING) ) { + copyStream(in, out) + } + } + } + finally { + ((HttpURLConnection)conn).disconnect() + } + } + + private static InputStream checkedInputStream(URLConnection conn, long expectedLength) throws IOException { + return expectedLength > 0 + ? new FixedInputStream(conn.getInputStream(), expectedLength) + : conn.getInputStream() + } + + /** + * Parse a {@code Content-Range} response header (e.g. {@code bytes 10-19/20}). + * + * @return an array {@code [start, end, total]}, or {@code null} when the header is absent or malformed + */ + private static long[] parseContentRange(URLConnection conn) { + final String value = conn.getHeaderField('Content-Range') + if( !value ) + return null + final matcher = value =~ ~/^bytes\s+(\d+)-(\d+)\/(\d+)$/ + return matcher.matches() + ? [matcher.group(1) as long, matcher.group(2) as long, matcher.group(3) as long] as long[] + : null + } + + private static void copyStream(InputStream in, OutputStream out) throws IOException { + final byte[] buffer = new byte[8192] + int len + while( (len = in.read(buffer)) != -1 ) + out.write(buffer, 0, len) + } + + @Override + void upload(Path source, Path target, CopyOption... options) throws IOException { + throw new UnsupportedOperationException("Upload not supported by ${getScheme().toUpperCase()} file system provider") + } + @Override boolean isSameFile(Path path, Path path2) throws IOException { return path == path2 diff --git a/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy b/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy index d9368c0c7b..b1a45e7dcc 100644 --- a/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy +++ b/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy @@ -16,10 +16,13 @@ package nextflow.file.http +import java.nio.file.FileAlreadyExistsException import java.nio.file.Files import java.nio.file.Path +import java.nio.file.StandardCopyOption import com.github.tomakehurst.wiremock.junit.WireMockRule +import nextflow.file.HttpCopyOption import org.junit.Rule import spock.lang.IgnoreIf import spock.lang.Specification @@ -228,4 +231,243 @@ class XFileSystemProviderTest extends Specification { 'this/that' | 'http://foo.com:123/abc' | 'http://foo.com:123/this/that' } + + def 'should resume an http download using a byte range'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' // 20 bytes + def PARTIAL = 'ABCDEFGHIJ' // first 10 bytes + def REMAINING = 'KLMNOPQRST' // remaining 10 bytes + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .withHeader('Range', equalTo('bytes=10-')) + .willReturn(aResponse() + .withStatus(206) + .withHeader('Content-Range', 'bytes 10-19/20') + .withHeader('Content-Length', '10') + .withBody(REMAINING))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-resume', '.txt') + target.text = PARTIAL + + when: + provider.download(source, target, HttpCopyOption.RESUME) + + then: + target.text == FULL + + and: + wireMockRule.verify(getRequestedFor(urlEqualTo('/file.txt')).withHeader('Range', equalTo('bytes=10-'))) + + cleanup: + target?.delete() + } + + def 'should restart an http download when the server ignores the range'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' + def PARTIAL = 'ABCDEFGHIJ' + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody(FULL))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-restart', '.txt') + target.text = PARTIAL + + when: + provider.download(source, target, HttpCopyOption.RESUME) + + then: + target.text == FULL + + and: + wireMockRule.verify(getRequestedFor(urlEqualTo('/file.txt')).withHeader('Range', equalTo('bytes=10-'))) + + cleanup: + target?.delete() + } + + def 'should gate resume download to http and https'() { + given: + def http = new HttpFileSystemProvider() + def https = new HttpsFileSystemProvider() + def ftp = new FtpFileSystemProvider() + + expect: + http.canDownload(null, null) + https.canDownload(null, null) + !ftp.canDownload(null, null) + + and: + !http.canUpload(null, null) + !https.canUpload(null, null) + !ftp.canUpload(null, null) + } + + def 'should restart when the server returns a mismatched content range'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' // 20 bytes + def PARTIAL = 'ABCDEFGHIJ' // first 10 bytes + + // ranged request returns a 206 that does not start at the requested offset + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .atPriority(1) + .withHeader('Range', equalTo('bytes=10-')) + .willReturn(aResponse() + .withStatus(206) + .withHeader('Content-Range', 'bytes 0-19/20') + .withHeader('Content-Length', '20') + .withBody(FULL))) + + // fallback request returns the full body + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .atPriority(2) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody(FULL))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-mismatch', '.txt') + target.text = PARTIAL + + when: + provider.download(source, target, HttpCopyOption.RESUME) + + then: + target.text == FULL + + and: + wireMockRule.verify(getRequestedFor(urlEqualTo('/file.txt')).withHeader('Range', equalTo('bytes=10-'))) + + cleanup: + target?.delete() + } + + def 'should restart when the server returns 416 range not satisfiable'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' + def PARTIAL = 'ABCDEFGHIJ' + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .atPriority(1) + .withHeader('Range', equalTo('bytes=10-')) + .willReturn(aResponse() + .withStatus(416) + .withHeader('Content-Range', 'bytes */20'))) + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .atPriority(2) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody(FULL))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-416', '.txt') + target.text = PARTIAL + + when: + provider.download(source, target, HttpCopyOption.RESUME) + + then: + target.text == FULL + + and: + wireMockRule.verify(getRequestedFor(urlEqualTo('/file.txt')).withHeader('Range', equalTo('bytes=10-'))) + + cleanup: + target?.delete() + } + + def 'should download a file when the target does not exist'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody(FULL))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-fresh', '.txt') + Files.delete(target) + + when: + provider.download(source, target) + + then: + target.text == FULL + + cleanup: + target?.delete() + } + + def 'should replace an existing file when REPLACE_EXISTING is specified'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody(FULL))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-replace', '.txt') + target.text = 'stale content' + + when: + provider.download(source, target, StandardCopyOption.REPLACE_EXISTING) + + then: + target.text == FULL + + cleanup: + target?.delete() + } + + def 'should fail when the target already exists without REPLACE_EXISTING'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def FULL = 'ABCDEFGHIJKLMNOPQRST' + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .willReturn(aResponse() + .withStatus(200) + .withHeader('Content-Length', '20') + .withBody(FULL))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def target = Files.createTempFile('nf-exists', '.txt') + target.text = 'stale content' + + when: + provider.download(source, target) + + then: + thrown(FileAlreadyExistsException) + target.text == 'stale content' + + cleanup: + target?.delete() + } } From 53fcea300d3c628812e69af48cf1c4f30c98f7db Mon Sep 17 00:00:00 2001 From: Stephen Watts Date: Sat, 15 Aug 2026 09:32:05 +1000 Subject: [PATCH 2/2] Support resume retries for AWS S3 destinations Assisted-by: Claude Code Signed-off-by: Stephen Watts --- .../groovy/nextflow/file/FilePorter.groovy | 41 +++++- .../nextflow/file/FilePorterTest.groovy | 60 ++++++++ .../nextflow/file/ResumableFileSystem.groovy | 42 ++++++ .../main/nextflow/file/ResumableUpload.groovy | 52 +++++++ .../file/http/XFileSystemProvider.groovy | 86 +++++++++++- .../file/http/XFileSystemProviderTest.groovy | 64 +++++++++ .../cloud/aws/nio/S3FileSystemProvider.java | 129 +++++++++++++++++- .../cloud/aws/nio/S3OutputStream.java | 74 +++++++++- .../aws/nio/S3FileSystemProviderTest.groovy | 123 +++++++++++++++++ .../cloud/aws/nio/S3OutputStreamTest.groovy | 102 ++++++++++++++ 10 files changed, 756 insertions(+), 17 deletions(-) create mode 100644 modules/nf-commons/src/main/nextflow/file/ResumableFileSystem.groovy create mode 100644 modules/nf-commons/src/main/nextflow/file/ResumableUpload.groovy diff --git a/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy b/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy index dd628ac53e..1675e3f060 100644 --- a/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy @@ -335,7 +335,7 @@ class FilePorter { } catch( IOException e ) { // check if a stage/download retry is allowed - final retry = count < maxRetries && !Thread.currentThread().isInterrupted() + final retry = count < maxRetries && recoverableError(e) && !Thread.currentThread().isInterrupted() // resume a partial download (HTTP/HTTPS only), otherwise discard the // partially downloaded file and start again from the beginning resume = retry && canResumeDownload(filePath, stagePath) @@ -352,6 +352,9 @@ class FilePorter { continue } + final targetProvider = stagePath.fileSystem.provider() + if( targetProvider instanceof ResumableFileSystem ) + abortResumableUpload((ResumableFileSystem)targetProvider, stagePath) throw new ProcessStageException(fmtError(filePath,e), e) } } @@ -359,13 +362,39 @@ class FilePorter { @PackageScope boolean canResumeDownload(Path source, Path target) { - // a resume only makes sense for a partially downloaded HTTP(S) file written to a local - // target, where APPEND is supported - if( !Files.exists(target) || Files.size(target) == 0 ) + if( source.toUri().scheme !in ['http', 'https'] ) return false - if( target.fileSystem != FileSystems.getDefault() ) + // local target: resume a partial file + if( target.fileSystem == FileSystems.getDefault() ) + return Files.exists(target) && Files.size(target) > 0 + // cloud target: resume an in-progress upload + final provider = target.fileSystem.provider() + if( provider !instanceof ResumableFileSystem ) return false - return source.toUri().scheme in ['http', 'https'] + try { + return ((ResumableFileSystem)provider).resumeUpload(target) != null + } + catch( IOException e ) { + log.debug "Unable to determine resume state for ${target.toUriString()}: ${e.message}" + return false + } + } + + private boolean recoverableError(IOException e) { + return e !instanceof NoSuchFileException + && (e instanceof SocketTimeoutException || e !instanceof InterruptedIOException); + } + + @PackageScope + void abortResumableUpload(ResumableFileSystem provider, Path target) { + try { + final upload = provider.resumeUpload(target) + if( upload != null ) + upload.abort() + } + catch( IOException e ) { + log.debug "Unable to abort in-progress upload for ${target.toUriString()}: ${e.message}" + } } private String fmtError(Path filePath, Exception e) { diff --git a/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy b/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy index 6917d16d05..ea2d061aa4 100644 --- a/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy @@ -18,6 +18,7 @@ package nextflow.file import java.nio.file.FileSystems import java.nio.file.Files +import java.nio.file.NoSuchFileException import java.nio.file.Path import java.util.concurrent.Semaphore @@ -158,6 +159,50 @@ class FilePorterTest extends Specification { } + def 'should not retry a missing source file' () { + given: + def source = 'http://localhost:1234/file.txt' as Path + def folder = Files.createTempDirectory('test') + def target = folder.resolve('file.txt') + + and: + def transfer = new NoSuchFileTransfer(source, target, 3) + + when: + transfer.stageForeignFile(source, target) + + then: + thrown(ProcessStageException) + transfer.attempts == 1 + + cleanup: + folder?.deleteDir() + } + + + def 'should abort an in-progress upload on terminal failure' () { + given: + def provider = Mock(ResumableFileSystem) + def upload = Mock(ResumableUpload) + def source = 'http://localhost:1234/file.txt' as Path + def folder = Files.createTempDirectory('test') + def target = folder.resolve('file.txt') + + and: + def transfer = new FilePorter.FileTransfer(source, target, 3, Mock(Semaphore)) + + when: + transfer.abortResumableUpload(provider, target) + + then: + 1 * provider.resumeUpload(target) >> upload + 1 * upload.abort() + + cleanup: + folder?.deleteDir() + } + + def 'should resume a partial download via the copyPath seam' () { given: @@ -273,6 +318,21 @@ class FilePorterTest extends Specification { } + static class NoSuchFileTransfer extends FilePorter.FileTransfer { + int attempts = 0 + + NoSuchFileTransfer(Path source, Path target, int maxRetries) { + super(source, target, maxRetries, new Semaphore(100)) + } + + @Override + Path copyForeignFile(Path source, Path target, boolean resume) { + attempts++ + throw new NoSuchFileException(source.toString()) + } + } + + def 'should copy foreign files' () { given: diff --git a/modules/nf-commons/src/main/nextflow/file/ResumableFileSystem.groovy b/modules/nf-commons/src/main/nextflow/file/ResumableFileSystem.groovy new file mode 100644 index 0000000000..a79b90ee60 --- /dev/null +++ b/modules/nf-commons/src/main/nextflow/file/ResumableFileSystem.groovy @@ -0,0 +1,42 @@ +/* + * Copyright 2013-2026, Seqera Labs + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package nextflow.file + +import java.io.IOException +import java.nio.file.CopyOption +import java.nio.file.Path + +/** + * A file system provider whose targets can resume an upload after an interruption. + * + * Implemented by providers that support resumable writes (e.g. S3 multipart upload). Used by a + * range-capable source provider to continue staging into a partially-uploaded target. + */ +interface ResumableFileSystem { + + /** + * Start a fresh upload of {@code target}, returning a handle that can write and complete it. + */ + ResumableUpload newUpload(Path target, CopyOption... options) throws IOException + + /** + * Recover an in-progress upload of {@code target}, returning a handle that can continue it, + * or {@code null} when there is no resumable in-progress upload. + */ + ResumableUpload resumeUpload(Path target) throws IOException + +} diff --git a/modules/nf-commons/src/main/nextflow/file/ResumableUpload.groovy b/modules/nf-commons/src/main/nextflow/file/ResumableUpload.groovy new file mode 100644 index 0000000000..2b0fdc8ae1 --- /dev/null +++ b/modules/nf-commons/src/main/nextflow/file/ResumableUpload.groovy @@ -0,0 +1,52 @@ +/* + * Copyright 2013-2026, Seqera Labs + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package nextflow.file + +import java.io.IOException +import java.io.OutputStream + +/** + * A handle to an in-progress resumable upload, produced by {@link ResumableFileSystem#resumeUpload}. + */ +interface ResumableUpload { + + /** + * @return The number of bytes already committed to the upload. + */ + long committedBytes() + + /** + * @return An output stream that writes the remaining bytes of the upload. + */ + OutputStream outputStream() + + /** + * Finish the upload, making the target object visible. + */ + void complete() throws IOException + + /** + * Abandon the upload, leaving it in-progress so a later attempt can resume it. + */ + void abandon() throws IOException + + /** + * Abort the upload, discarding the in-progress upload entirely. + */ + void abort() throws IOException + +} diff --git a/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy b/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy index f5a8558873..74bcd5e463 100644 --- a/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy +++ b/modules/nf-httpfs/src/main/nextflow/file/http/XFileSystemProvider.groovy @@ -20,6 +20,8 @@ import nextflow.file.CopyMoveHelper import nextflow.file.CopyOptions import nextflow.file.FileSystemTransferAware import nextflow.file.HttpCopyOption +import nextflow.file.ResumableFileSystem +import nextflow.file.ResumableUpload import static nextflow.file.http.XFileSystemConfig.* @@ -450,7 +452,18 @@ abstract class XFileSystemProvider extends FileSystemProvider implements FileSys throw new ProviderMismatchException() if( options.contains(HttpCopyOption.RESUME) ) { - // resume a partially downloaded file, falling back to a full download + // resume into a resumable (cloud) target when available + final targetProvider = target.fileSystem.provider() + if( targetProvider instanceof ResumableFileSystem ) { + final upload = ((ResumableFileSystem)targetProvider).resumeUpload(target) + if( upload != null ) { + if( upload.committedBytes() > 0 && resumeToTarget(source, upload) ) + return + // nothing committed to resume from: discard the stale upload and restart + upload.abort() + } + } + // resume into a local target, otherwise fall back to a full download final long offset = Files.exists(target) ? Files.size(target) : 0 if( offset > 0 && resumeDownload(source, target, offset) ) return @@ -498,6 +511,47 @@ abstract class XFileSystemProvider extends FileSystemProvider implements FileSys } } + /** + * Resume a download into a resumable (cloud) target from the committed offset. + * + * @return {@code true} when the remaining bytes were streamed and the upload completed, + * {@code false} when the source could not be resumed (and the upload was aborted) + */ + @PackageScope + boolean resumeToTarget(Path source, ResumableUpload upload) throws IOException { + final long offset = upload.committedBytes() + final conn = toConnection(source, offset) + if( conn !instanceof HttpURLConnection ) { + upload.abort() + return false + } + try { + final int code = ((HttpURLConnection)conn).getResponseCode() + final long[] range = code == 206 ? parseContentRange(conn) : null + if( range == null || range[0] != offset || range[1] + 1 != range[2] ) { + // source can't be resumed — discard the stale upload and restart + upload.abort() + return false + } + final long remaining = range[2] - offset + try { + try( InputStream in = checkedInputStream(conn, remaining) ) { + copyStream(in, upload.outputStream()) + } + upload.complete() + } + catch( IOException e ) { + // mid-stream failure — leave the upload in-progress for a later retry + upload.abandon() + throw e + } + return true + } + finally { + ((HttpURLConnection)conn).disconnect() + } + } + private void downloadFromStart(Path source, Path target) throws IOException { final conn = toConnection(source) if( conn !instanceof HttpURLConnection ) @@ -506,9 +560,15 @@ abstract class XFileSystemProvider extends FileSystemProvider implements FileSys final int code = ((HttpURLConnection)conn).getResponseCode() if( code != 200 ) throw new IOException("Unable to download foreign file ${FilesEx.toUriString(source)} -- unexpected HTTP status code: $code") - try( InputStream in = checkedInputStream(conn, conn.getContentLengthLong()) ) { - try( OutputStream out = Files.newOutputStream(target, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING) ) { - copyStream(in, out) + final targetProvider = target.fileSystem.provider() + if( targetProvider instanceof ResumableFileSystem ) { + downloadToResumableTarget(conn, ((ResumableFileSystem)targetProvider).newUpload(target)) + } + else { + try( InputStream in = checkedInputStream(conn, conn.getContentLengthLong()) ) { + try( OutputStream out = Files.newOutputStream(target, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING) ) { + copyStream(in, out) + } } } } @@ -517,6 +577,24 @@ abstract class XFileSystemProvider extends FileSystemProvider implements FileSys } } + /** + * Stream a download into a resumable (cloud) target, completing the upload on success and + * leaving it in-progress on a mid-stream failure so a later attempt can resume it. + */ + private void downloadToResumableTarget(URLConnection conn, ResumableUpload upload) throws IOException { + try { + try( InputStream in = checkedInputStream(conn, conn.getContentLengthLong()) ) { + copyStream(in, upload.outputStream()) + } + upload.complete() + } + catch( IOException e ) { + // mid-stream failure — leave the upload in-progress for a later retry + upload.abandon() + throw e + } + } + private static InputStream checkedInputStream(URLConnection conn, long expectedLength) throws IOException { return expectedLength > 0 ? new FixedInputStream(conn.getInputStream(), expectedLength) diff --git a/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy b/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy index b1a45e7dcc..e7adcc43cb 100644 --- a/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy +++ b/modules/nf-httpfs/src/test/nextflow/file/http/XFileSystemProviderTest.groovy @@ -23,6 +23,7 @@ import java.nio.file.StandardCopyOption import com.github.tomakehurst.wiremock.junit.WireMockRule import nextflow.file.HttpCopyOption +import nextflow.file.ResumableUpload import org.junit.Rule import spock.lang.IgnoreIf import spock.lang.Specification @@ -470,4 +471,67 @@ class XFileSystemProviderTest extends Specification { cleanup: target?.delete() } + + def 'should resume a download into a resumable upload'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + def REMAINING = 'KLMNOPQRST' + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .withHeader('Range', equalTo('bytes=10-')) + .willReturn(aResponse() + .withStatus(206) + .withHeader('Content-Range', 'bytes 10-19/20') + .withHeader('Content-Length', '10') + .withBody(REMAINING))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def upload = Mock(ResumableUpload) + def output = new ByteArrayOutputStream() + + when: + def resumed = provider.resumeToTarget(source, upload) + + then: + 1 * upload.committedBytes() >> 10 + 1 * upload.outputStream() >> output + 1 * upload.complete() + 0 * upload.abort() + 0 * upload.abandon() + + and: + resumed + output.toString() == REMAINING + } + + def 'should abandon a resumable upload when the download fails mid-stream'() { + given: + def localhost = "http://localhost:${wireMockRule.port()}" + + wireMockRule.stubFor(get(urlEqualTo('/file.txt')) + .withHeader('Range', equalTo('bytes=10-')) + .willReturn(aResponse() + .withStatus(206) + .withHeader('Content-Range', 'bytes 10-19/20') + .withHeader('Content-Length', '10') + .withBody('KLMNOPQRST'))) + + def provider = new HttpFileSystemProvider() + def source = provider.getPath(new URI("${localhost}/file.txt")) + def upload = Mock(ResumableUpload) + def failingOutput = new OutputStream() { + void write(int b) throws IOException { throw new IOException('boom') } + } + + when: + provider.resumeToTarget(source, upload) + + then: + 1 * upload.committedBytes() >> 10 + 1 * upload.outputStream() >> failingOutput + 1 * upload.abandon() + 0 * upload.complete() + thrown(IOException) + } } diff --git a/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3FileSystemProvider.java b/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3FileSystemProvider.java index dfc1c9bf0d..0f48a5c068 100644 --- a/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3FileSystemProvider.java +++ b/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3FileSystemProvider.java @@ -48,6 +48,7 @@ import java.nio.file.attribute.FileAttributeView; import java.nio.file.attribute.FileTime; import java.nio.file.spi.FileSystemProvider; +import java.util.ArrayList; import java.util.Arrays; import java.util.EnumSet; import java.util.HashMap; @@ -61,6 +62,7 @@ import java.util.concurrent.TimeUnit; import software.amazon.awssdk.core.ResponseInputStream; +import software.amazon.awssdk.core.exception.SdkException; import software.amazon.awssdk.services.s3.model.*; import software.amazon.awssdk.services.s3.model.S3Object; import com.google.common.base.Preconditions; @@ -77,6 +79,8 @@ import nextflow.file.CopyOptions; import nextflow.file.FileHelper; import nextflow.file.FileSystemTransferAware; +import nextflow.file.ResumableFileSystem; +import nextflow.file.ResumableUpload; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -109,7 +113,7 @@ * * */ -public class S3FileSystemProvider extends FileSystemProvider implements FileSystemTransferAware { +public class S3FileSystemProvider extends FileSystemProvider implements FileSystemTransferAware, ResumableFileSystem { private static final Logger log = LoggerFactory.getLogger(S3FileSystemProvider.class); @@ -345,20 +349,139 @@ else if ( exits ) } private S3OutputStream createUploaderOutputStream( S3Path fileToUpload ) { + return createUploaderOutputStream(fileToUpload, null, null); + } + + private S3OutputStream createUploaderOutputStream( S3Path fileToUpload, String uploadId, List completedParts ) { S3Client s3 = fileToUpload.getFileSystem().getClient(); Properties props = fileToUpload.getFileSystem().properties(); final String storageClass = fileToUpload.getStorageClass()!=null ? fileToUpload.getStorageClass() : props.getProperty("upload_storage_class"); final S3MultipartOptions opts = props != null ? new S3MultipartOptions(props) : new S3MultipartOptions(); final S3ObjectId objectId = fileToUpload.toS3ObjectId(); - S3OutputStream stream = new S3OutputStream(s3.getClient(), objectId, opts) + final S3OutputStream stream = uploadId == null + ? new S3OutputStream(s3.getClient(), objectId, opts) + : new S3OutputStream(s3.getClient(), objectId, opts, uploadId, completedParts); + return stream .setCannedAcl(s3.getCannedAcl()) .setStorageClass(storageClass) .setStorageEncryption(props.getProperty("storage_encryption")) .setKmsKeyId(props.getProperty("storage_kms_key_id")) .setContentType(fileToUpload.getContentType()) .setTags(fileToUpload.getTagsList()); - return stream; + } + + @Override + public ResumableUpload newUpload(Path target, CopyOption... options) throws IOException { + if( !(target instanceof S3Path) ) + return null; + final S3Path s3Path = (S3Path)target; + final S3OutputStream stream = createUploaderOutputStream(s3Path, null, null); + return new S3ResumableUpload(stream, 0); + } + + @Override + public ResumableUpload resumeUpload(Path target) throws IOException { + if( !(target instanceof S3Path) ) + return null; + final S3Path s3Path = (S3Path)target; + final software.amazon.awssdk.services.s3.S3Client awsClient = s3Path.getFileSystem().getClient().getClient(); + final String bucket = s3Path.getBucket(); + final String key = s3Path.getKey(); + + // find the most recent in-progress multipart upload for this object, paging through all + // results since listMultipartUploads returns at most 1000 entries per call + MultipartUpload upload = null; + String keyMarker = null; + String uploadIdMarker = null; + ListMultipartUploadsResponse listResp; + do { + try { + listResp = awsClient.listMultipartUploads( + ListMultipartUploadsRequest.builder() + .bucket(bucket) + .prefix(key) + .keyMarker(keyMarker) + .uploadIdMarker(uploadIdMarker) + .build()); + } + catch( final SdkException e ) { + throw new IOException("Failed to list Amazon S3 multipart uploads", e); + } + for( final MultipartUpload candidate : listResp.uploads() ) { + if( key.equals(candidate.key()) && (upload == null || candidate.initiated().isAfter(upload.initiated())) ) + upload = candidate; + } + keyMarker = listResp.nextKeyMarker(); + uploadIdMarker = listResp.nextUploadIdMarker(); + } while( Boolean.TRUE.equals(listResp.isTruncated()) ); + + if( upload == null ) + return null; + + // recover the already-uploaded parts, paging through all results since listParts returns + // at most 1000 entries per call + final List completedParts = new ArrayList<>(); + long committedBytes = 0; + Integer partNumberMarker = null; + ListPartsResponse partsResp; + do { + try { + partsResp = awsClient.listParts( + ListPartsRequest.builder() + .bucket(bucket) + .key(key) + .uploadId(upload.uploadId()) + .partNumberMarker(partNumberMarker) + .build()); + } + catch( final SdkException e ) { + throw new IOException("Failed to list Amazon S3 multipart upload parts", e); + } + for( final Part part : partsResp.parts() ) { + completedParts.add(CompletedPart.builder().partNumber(part.partNumber()).eTag(part.eTag()).build()); + committedBytes += part.size(); + } + partNumberMarker = partsResp.nextPartNumberMarker(); + } while( Boolean.TRUE.equals(partsResp.isTruncated()) ); + + final S3OutputStream stream = createUploaderOutputStream(s3Path, upload.uploadId(), completedParts); + return new S3ResumableUpload(stream, committedBytes); + } + + private static class S3ResumableUpload implements ResumableUpload { + private final S3OutputStream stream; + private final long committedBytes; + + S3ResumableUpload(S3OutputStream stream, long committedBytes) { + this.stream = stream; + this.committedBytes = committedBytes; + } + + @Override + public long committedBytes() { + return committedBytes; + } + + @Override + public OutputStream outputStream() { + return stream; + } + + @Override + public void complete() throws IOException { + stream.close(); + } + + @Override + public void abandon() throws IOException { + stream.abandon(); + } + + @Override + public void abort() throws IOException { + stream.abort(); + } } @Override diff --git a/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3OutputStream.java b/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3OutputStream.java index a945f83c6d..79aa88979c 100644 --- a/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3OutputStream.java +++ b/plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3OutputStream.java @@ -160,6 +160,22 @@ public S3OutputStream(final S3Client s3, S3ObjectId objectId, S3MultipartOptions this.bufferSize = request.getBufferSize(); } + /** + * Creates an {@code S3OutputStream} that continues an existing multipart upload identified by + * {@code uploadId}, resuming part numbering after the already-completed {@code completedParts}. + */ + public S3OutputStream(final S3Client s3, S3ObjectId objectId, S3MultipartOptions request, + String uploadId, List completedParts) { + this.s3 = requireNonNull(s3); + this.objectId = requireNonNull(objectId); + this.request = request; + this.bufferSize = request.getBufferSize(); + this.uploadId = requireNonNull(uploadId); + this.completedParts = new LinkedBlockingQueue<>(completedParts); + this.partsCount = completedParts.size(); + // executor and phaser are created lazily on the first upload (see uploadBuffer) + } + private ByteBuffer expandBuffer(ByteBuffer byteBuffer) { final float expandFactor = 2.5f; @@ -306,9 +322,16 @@ private boolean uploadBuffer(ByteBuffer buf, boolean last) throws IOException { return false; } - if (partsCount == 0) { + if (uploadId == null) { init(); } + else if (executor == null) { + // a resumed upload already has an uploadId and completed parts, so only the + // background executor and phaser need to be created + executor = getOrCreateExecutor(request.getMaxThreads()); + phaser = new Phaser(); + phaser.register(); + } // set the buffer in read mode and submit for upload executor.submit( task(buf, md5.digest(), ++partsCount) ); @@ -388,12 +411,16 @@ public void close() throws IOException { } else { // -- upload remaining chunk - if( buf != null ) + if( buf != null ) { uploadBuffer(buf, true); + buf = null; + md5 = null; + } // -- shutdown upload executor and await termination log.trace("[S3 phaser] Close arriveAndAwaitAdvance"); - phaser.arriveAndAwaitAdvance(); + if( phaser != null ) + phaser.arriveAndAwaitAdvance(); // -- complete upload process completeMultipartUpload(); @@ -402,6 +429,44 @@ public void close() throws IOException { closed = true; } + /** + * Abandon the upload, leaving it in-progress so a later attempt can resume it. Flushes any + * buffered data that forms a full minimum part and waits for the background parts, but does + * not complete or abort the multipart upload. + */ + public void abandon() throws IOException { + if( closed ) + return; + + if( uploadId != null ) { + // upload any remaining buffered data as a part; a tail smaller than the minimum part + // size is left uncommitted, since a resume would make it a middle part and S3 would + // reject the completion with EntityTooSmall + if( buf != null && buf.position() >= MIN_MULTIPART_UPLOAD ) { + uploadBuffer(buf, true); + buf = null; + md5 = null; + } + // wait for the background part uploads to complete + if( phaser != null ) + phaser.arriveAndAwaitAdvance(); + // do NOT complete and do NOT abort — leave the multipart upload in-progress + } + + closed = true; + } + + /** + * Abort the upload, discarding the in-progress multipart upload entirely. + */ + public void abort() { + if( closed ) + return; + if( uploadId != null ) + abortMultipartUpload(); + closed = true; + } + /** * Starts the multipart upload process * @@ -534,7 +599,8 @@ private synchronized void abortMultipartUpload() { } aborted = true; log.trace("[S3 phaser] MultipartUpload arriveAndDeregister"); - phaser.arriveAndDeregister(); + if( phaser != null ) + phaser.arriveAndDeregister(); } /** diff --git a/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3FileSystemProviderTest.groovy b/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3FileSystemProviderTest.groovy index dd83e9ada6..9681c7cef0 100644 --- a/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3FileSystemProviderTest.groovy +++ b/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3FileSystemProviderTest.groovy @@ -16,7 +16,16 @@ package nextflow.cloud.aws.nio +import java.time.Instant + +import software.amazon.awssdk.services.s3.model.ListMultipartUploadsRequest +import software.amazon.awssdk.services.s3.model.ListMultipartUploadsResponse +import software.amazon.awssdk.services.s3.model.ListPartsRequest +import software.amazon.awssdk.services.s3.model.ListPartsResponse +import software.amazon.awssdk.services.s3.model.MultipartUpload import software.amazon.awssdk.services.s3.model.ObjectCannedACL +import software.amazon.awssdk.services.s3.model.Part +import software.amazon.awssdk.services.s3.model.S3Exception import software.amazon.awssdk.services.s3.model.ServerSideEncryption import spock.lang.Specification @@ -95,5 +104,119 @@ class S3FileSystemProviderTest extends Specification { fs.properties().getProperty('multipart_threshold') == '33554432' //32MB } + def 'should resume an in-progress multipart upload'() { + given: + def awsClient = Mock(software.amazon.awssdk.services.s3.S3Client) + def nfClient = Mock(nextflow.cloud.aws.nio.S3Client) { + getClient() >> awsClient + } + def fs = Mock(S3FileSystem) { + getClient() >> nfClient + properties() >> new Properties() + } + def s3Path = new S3Path(fs, '/bucket/key.txt') + def provider = new S3FileSystemProvider() + + def upload = MultipartUpload.builder().key('key.txt').uploadId('upload-id').initiated(Instant.now()).build() + def part1 = Part.builder().partNumber(1).size(10).eTag('etag1').build() + def part2 = Part.builder().partNumber(2).size(20).eTag('etag2').build() + + when: + def handle = provider.resumeUpload(s3Path) + + then: + 1 * awsClient.listMultipartUploads(_) >> ListMultipartUploadsResponse.builder().uploads([upload]).build() + 1 * awsClient.listParts(_) >> ListPartsResponse.builder().parts([part1, part2]).build() + + and: + handle != null + handle.committedBytes() == 30 + handle.outputStream() instanceof S3OutputStream + } + def 'should paginate listParts when resuming a multipart upload'() { + given: + def awsClient = Mock(software.amazon.awssdk.services.s3.S3Client) + def nfClient = Mock(nextflow.cloud.aws.nio.S3Client) { + getClient() >> awsClient + } + def fs = Mock(S3FileSystem) { + getClient() >> nfClient + properties() >> new Properties() + } + def s3Path = new S3Path(fs, '/bucket/key.txt') + def provider = new S3FileSystemProvider() + + def upload = MultipartUpload.builder().key('key.txt').uploadId('upload-id').initiated(Instant.now()).build() + def part1 = Part.builder().partNumber(1).size(10).eTag('etag1').build() + def part2 = Part.builder().partNumber(2).size(20).eTag('etag2').build() + + when: + def handle = provider.resumeUpload(s3Path) + + then: + 1 * awsClient.listMultipartUploads(_) >> ListMultipartUploadsResponse.builder().uploads([upload]).build() + 2 * awsClient.listParts(_) >> { ListPartsRequest req -> + req.partNumberMarker() == null + ? ListPartsResponse.builder().parts([part1]).isTruncated(true).nextPartNumberMarker(1).build() + : ListPartsResponse.builder().parts([part2]).isTruncated(false).build() + } + + and: + handle != null + handle.committedBytes() == 30 + } + + def 'should paginate listMultipartUploads when resuming'() { + given: + def awsClient = Mock(software.amazon.awssdk.services.s3.S3Client) + def nfClient = Mock(nextflow.cloud.aws.nio.S3Client) { + getClient() >> awsClient + } + def fs = Mock(S3FileSystem) { + getClient() >> nfClient + properties() >> new Properties() + } + def s3Path = new S3Path(fs, '/bucket/key.txt') + def provider = new S3FileSystemProvider() + + def upload = MultipartUpload.builder().key('key.txt').uploadId('upload-id').initiated(Instant.now()).build() + def part1 = Part.builder().partNumber(1).size(10).eTag('etag1').build() + + when: + def handle = provider.resumeUpload(s3Path) + + then: + 2 * awsClient.listMultipartUploads(_) >> { ListMultipartUploadsRequest req -> + req.keyMarker() == null + ? ListMultipartUploadsResponse.builder().uploads([]).isTruncated(true).nextKeyMarker('key.txt').build() + : ListMultipartUploadsResponse.builder().uploads([upload]).isTruncated(false).build() + } + 1 * awsClient.listParts(_) >> ListPartsResponse.builder().parts([part1]).build() + + and: + handle != null + handle.committedBytes() == 10 + } + + def 'should propagate S3 errors when resuming'() { + given: + def awsClient = Mock(software.amazon.awssdk.services.s3.S3Client) + def nfClient = Mock(nextflow.cloud.aws.nio.S3Client) { + getClient() >> awsClient + } + def fs = Mock(S3FileSystem) { + getClient() >> nfClient + properties() >> new Properties() + } + def s3Path = new S3Path(fs, '/bucket/key.txt') + def provider = new S3FileSystemProvider() + + when: + provider.resumeUpload(s3Path) + + then: + 1 * awsClient.listMultipartUploads(_) >> { throw S3Exception.builder().statusCode(503).build() } + thrown(IOException) + } } diff --git a/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3OutputStreamTest.groovy b/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3OutputStreamTest.groovy index 4de1865bff..279cf75a43 100644 --- a/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3OutputStreamTest.groovy +++ b/plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/S3OutputStreamTest.groovy @@ -21,8 +21,10 @@ import nextflow.Session import nextflow.cloud.aws.nio.util.S3MultipartOptions import nextflow.file.FileHelper import software.amazon.awssdk.services.s3.S3Client +import software.amazon.awssdk.services.s3.model.CompletedPart import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse +import software.amazon.awssdk.services.s3.model.S3Exception import software.amazon.awssdk.services.s3.model.UploadPartResponse import spock.lang.IgnoreIf import spock.lang.Requires @@ -158,4 +160,104 @@ class S3OutputStreamTest extends Specification implements AwsS3BaseSpec { capturedParts[2].partNumber() == 2 } + + def 'should resume a multipart upload'() { + given: + final path = s3path("s3://test/file.txt") + def multipart = new S3MultipartOptions() + def client = Mock(S3Client) + def recovered = [ + CompletedPart.builder().partNumber(1).eTag('etag1').build(), + CompletedPart.builder().partNumber(2).eTag('etag2').build(), + ] + + def writer = new S3OutputStream(client, path.toS3ObjectId(), multipart, 'upload-id', recovered) + + when: 'continue uploading from the recovered state' + writer.uploadPart(InputStream.nullInputStream(), 25, 'checksum'.bytes, 3, true) + writer.completeMultipartUpload() + + then: 'no new upload is initiated and the recovered parts are included' + 0 * client.createMultipartUpload(_) + 1 * client.uploadPart(_, _) >> { UploadPartResponse.builder().eTag('etag3').build() } + 1 * client.completeMultipartUpload(_) >> { CompleteMultipartUploadRequest req -> + assert req.uploadId() == 'upload-id' + assert req.multipartUpload().parts()*.eTag() == ['etag1', 'etag2', 'etag3'] + return null + } + } + + def 'should leave a sub-minimum buffer uncommitted on abandon'() { + given: + final path = s3path('s3://test/file.txt') + def multipart = new S3MultipartOptions() + def client = Mock(S3Client) + def writer = new S3OutputStream(client, path.toS3ObjectId(), multipart) + + when: + writer.init() + writer.write(new byte[1024]) + writer.abandon() + + then: + 1 * client.createMultipartUpload(_) >> CreateMultipartUploadResponse.builder().uploadId('upload-id').build() + writer.partsCount == 0 + 0 * client.uploadPart(_, _) + } + + def 'should not fail to abandon a resumed stream with no new bytes'() { + given: + final path = s3path('s3://test/file.txt') + def multipart = new S3MultipartOptions() + def client = Mock(S3Client) + def recovered = [CompletedPart.builder().partNumber(1).eTag('etag1').build()] + def writer = new S3OutputStream(client, path.toS3ObjectId(), multipart, 'upload-id', recovered) + + when: + writer.abandon() + + then: + noExceptionThrown() + } + + def 'should not fail to close a resumed stream with no new bytes'() { + given: + final path = s3path('s3://test/file.txt') + def multipart = new S3MultipartOptions() + def client = Mock(S3Client) + def recovered = [CompletedPart.builder().partNumber(1).eTag('etag1').build()] + def writer = new S3OutputStream(client, path.toS3ObjectId(), multipart, 'upload-id', recovered) + + when: + writer.close() + + then: + 1 * client.completeMultipartUpload(_) + } + + def 'should not re-upload the tail when abandoning after a failed close'() { + given: + final path = s3path('s3://test/file.txt') + def multipart = new S3MultipartOptions() + multipart.setBufferSize(5 * 1024 * 1024) + def client = Mock(S3Client) + def writer = new S3OutputStream(client, path.toS3ObjectId(), multipart) + + when: + writer.write(new byte[5 * 1024 * 1024]) + writer.flush() + writer.write(new byte[5 * 1024 * 1024]) + try { + writer.close() + } + catch( IOException e ) { + writer.abandon() + } + + then: + 1 * client.createMultipartUpload(_) >> CreateMultipartUploadResponse.builder().uploadId('upload-id').build() + 2 * client.uploadPart(_, _) >> { UploadPartResponse.builder().eTag('etag').build() } + 1 * client.completeMultipartUpload(_) >> { throw S3Exception.builder().statusCode(500).build() } + 0 * client.abortMultipartUpload(_) + } }