diff --git a/docs/reference/env-vars.md b/docs/reference/env-vars.md index 2c6210effa..fcbc79ea3f 100644 --- a/docs/reference/env-vars.md +++ b/docs/reference/env-vars.md @@ -161,6 +161,11 @@ The following environment variables control the configuration of the Nextflow ru `NXF_PLUGINS_DIR` : The path where the plugin archives are loaded and stored (default: `$NXF_HOME/plugins`). +`NXF_PLUGINS_REGISTRY_URL` +: :::{versionadded} 25.08.0-edge + ::: +: Specifies the URL of the plugin registry used to download and resolve plugins. This allows using custom or private plugin registries instead of the default public registry. + `NXF_PLUGINS_TEST_REPOSITORY` : :::{versionadded} 23.04.0 ::: @@ -191,6 +196,11 @@ The following environment variables control the configuration of the Nextflow ru ::: : Max delay used for HTTP retryable operations (default: `90s`). +`NXF_RETRY_POLICY_MULTIPLIER` +: :::{versionadded} 25.08.0-edge + ::: +: Delay multiplier used for HTTP retryable operations (default: `2.0`). + `NXF_SCM_FILE` : :::{versionadded} 20.10.0 ::: diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy index b82e227e60..1809f848fe 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy @@ -235,7 +235,7 @@ class PublishDir { protected void apply0(Set files) { assert path // setup the retry policy config to be used - this.retryConfig = RetryConfig.config(session) + this.retryConfig = RetryConfig.config(session.config) createPublishDir() validatePublishMode() diff --git a/modules/nextflow/src/main/groovy/nextflow/util/ConfigHelper.groovy b/modules/nextflow/src/main/groovy/nextflow/util/ConfigHelper.groovy index 6c211673af..e0c7222dfc 100644 --- a/modules/nextflow/src/main/groovy/nextflow/util/ConfigHelper.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/util/ConfigHelper.groovy @@ -1,5 +1,5 @@ /* - * Copyright 2013-2024, Seqera Labs + * Copyright 2013-2025, Seqera Labs * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,13 +18,10 @@ package nextflow.util import java.nio.file.Path -import com.google.common.base.CaseFormat import groovy.json.JsonOutput -import groovy.transform.CompileDynamic import groovy.transform.CompileStatic import groovy.transform.PackageScope import groovy.util.logging.Slf4j -import nextflow.SysEnv import nextflow.config.ConfigClosurePlaceholder import org.codehaus.groovy.runtime.InvokerHelper import org.yaml.snakeyaml.DumperOptions @@ -370,37 +367,5 @@ class ConfigHelper { return value } - static T valueOf(Map config, String name, String prefix, T defValue, Class type) { - assert name, "Argument 'name' cannot be null or empty" - assert type, "Argument 'type' cannot be null" - - // try to get the value from the config map - final cfg = config?.get(name) - if( cfg != null ) { - return toType(cfg, type) - } - // try to fallback to the sys environment - if( !prefix.endsWith('_') ) - prefix += '_' - final key = prefix.toUpperCase() + CaseFormat.LOWER_CAMEL.to(CaseFormat.UPPER_UNDERSCORE, name) - final env = SysEnv.get(key) - if( env != null ) { - return toType(env, type) - } - // return the default value - return defValue - } - - @CompileDynamic - static protected T toType(Object value, Class type) { - if( value == null ) - return null - if( type==Boolean.class ) { - return type.cast(Boolean.valueOf(value.toString())) - } - else { - return value.asType(type) - } - } } diff --git a/modules/nextflow/src/main/groovy/nextflow/util/RetryConfig.groovy b/modules/nextflow/src/main/groovy/nextflow/util/RetryConfig.groovy deleted file mode 100644 index ba6ec54b21..0000000000 --- a/modules/nextflow/src/main/groovy/nextflow/util/RetryConfig.groovy +++ /dev/null @@ -1,87 +0,0 @@ -/* - * Copyright 2013-2024, 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.util - -import static nextflow.util.ConfigHelper.* - -import groovy.transform.CompileStatic -import groovy.transform.EqualsAndHashCode -import groovy.transform.ToString -import groovy.util.logging.Slf4j -import nextflow.Global -import nextflow.Session -/** - * Models retry policy configuration - * - * @author Paolo Di Tommaso - */ -@Slf4j -@ToString(includePackage = false, includeNames = true) -@EqualsAndHashCode -@CompileStatic -class RetryConfig { - - private final static Duration DEFAULT_DELAY = Duration.of('350ms') - private final static Duration DEFAULT_MAX_DELAY = Duration.of('90s') - private final static Integer DEFAULT_MAX_ATTEMPTS = 5 - private final static Double DEFAULT_JITTER = 0.25 - - private final static String ENV_PREFIX = 'NXF_RETRY_POLICY_' - - final private Duration delay - final private Duration maxDelay - final private int maxAttempts - final private double jitter - - RetryConfig() { - this(Collections.emptyMap()) - } - - RetryConfig(Map config) { - delay = - valueOf(config, 'delay', ENV_PREFIX, DEFAULT_DELAY, Duration) - maxDelay = - valueOf(config, 'maxDelay', ENV_PREFIX, DEFAULT_MAX_DELAY, Duration) - maxAttempts = - valueOf(config, 'maxAttempts', ENV_PREFIX, DEFAULT_MAX_ATTEMPTS, Integer) - jitter = - valueOf(config, 'jitter', ENV_PREFIX, DEFAULT_JITTER, Double) - } - - Duration getDelay() { delay } - - Duration getMaxDelay() { maxDelay } - - int getMaxAttempts() { maxAttempts } - - double getJitter() { jitter } - - static RetryConfig config() { - config(Global.session as Session) - } - - static RetryConfig config(Session session) { - if( session ) { - return new RetryConfig(session.config.navigate('nextflow.retryPolicy') as Map ?: Collections.emptyMap()) - } - log.warn "Missing nextflow session - using default retry config" - return new RetryConfig() - } - - -} diff --git a/modules/nextflow/src/test/groovy/nextflow/util/ConfigHelperTest.groovy b/modules/nextflow/src/test/groovy/nextflow/util/ConfigHelperTest.groovy index d80274fd60..c1f1784e86 100644 --- a/modules/nextflow/src/test/groovy/nextflow/util/ConfigHelperTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/util/ConfigHelperTest.groovy @@ -19,11 +19,9 @@ package nextflow.util import java.nio.file.Files import java.nio.file.Paths -import nextflow.SysEnv import nextflow.config.ConfigClosurePlaceholder import spock.lang.Specification import spock.lang.Unroll - /** * * @author Paolo Di Tommaso @@ -354,72 +352,5 @@ class ConfigHelperTest extends Specification { "withName:2foo" | "'withName:2foo'" | "withName:'2foo'" } - @Unroll - def 'should get config from map' () { - given: - def NAME = 'foo' - def PREFIX = 'P_' - and: - SysEnv.push(ENV) - - expect: - ConfigHelper.valueOf(CONFIG, NAME, PREFIX, DEF_VAL, DEF_TYPE) == EXPECTED - - cleanup: - SysEnv.pop() - - where: - CONFIG | ENV | DEF_VAL | DEF_TYPE | EXPECTED - null | [:] | null | String | null - [:] | [:] | null | String | null - [:] | [:] | 'one' | String | 'one' - [foo:'two'] | [:] | 'one' | String | 'two' - [foo:''] | [:] | 'one' | String | '' - [foo:'two'] | [P_FOO:'bar'] | 'one' | String | 'two' - [:] | [P_FOO:'bar'] | 'one' | String | 'bar' - - and: - null | [:] | null | Integer | null - [:] | [:] | null | Integer | null - [:] | [:] | 1 | Integer | 1 - [foo:2] | [:] | 1 | Integer | 2 - [foo:'2'] | [:] | 1 | Integer | 2 - [foo:'2'] | [P_FOO:'3'] | 1 | Integer | 2 - [:] | [P_FOO:'3'] | 1 | Integer | 3 - - and: - null | [:] | null | Boolean | null - [:] | [:] | true | Boolean | true - [foo:false] | [:] | true | Boolean | false - [foo:'false'] | [:] | true | Boolean | false - [foo:true] | [:] | false | Boolean | true - [foo:'true'] | [:] | false | Boolean | true - [foo:'true'] | [P_FOO:'false']| null | Boolean | true - [:] | [P_FOO:'false']| null | Boolean | false - [:] | [P_FOO:'true'] | null | Boolean | true - - and: - [:] | [:] | Duration.of('1s') | Duration | Duration.of('1s') - [foo:'10ms'] | [:] | null | Duration | Duration.of('10ms') - [:] | [P_FOO:'1s'] | null | Duration | Duration.of('1s') - } - - def 'should map camelCase to snake uppercase' () { - given: - SysEnv.push(ENV) - - expect: - ConfigHelper.valueOf([:], NAME, PREFIX, null, String) == EXPECTED - - cleanup: - SysEnv.pop() - - where: - EXPECTED | PREFIX | NAME | ENV - null | 'foo' | 'bar' | [:] - 'one' | 'foo' | 'bar' | [FOO_BAR: 'one'] - 'one' | 'foo_' | 'bar' | [FOO_BAR: 'one'] - 'one' | 'foo_' | 'thisAndThat' | [FOO_THIS_AND_THAT: 'one'] - } } diff --git a/modules/nextflow/src/test/groovy/nextflow/util/RetryConfigTest.groovy b/modules/nextflow/src/test/groovy/nextflow/util/RetryConfigTest.groovy deleted file mode 100644 index db99cf343c..0000000000 --- a/modules/nextflow/src/test/groovy/nextflow/util/RetryConfigTest.groovy +++ /dev/null @@ -1,63 +0,0 @@ -/* - * Copyright 2013-2024, 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.util - -import nextflow.SysEnv -import spock.lang.Specification -/** - * - * @author Ben Sherman - */ -class RetryConfigTest extends Specification { - - def 'should create retry config' () { - - expect: - new RetryConfig().delay == Duration.of('350ms') - new RetryConfig().maxDelay == Duration.of('90s') - new RetryConfig().maxAttempts == 5 - new RetryConfig().jitter == 0.25d - - and: - new RetryConfig([maxAttempts: 20]).maxAttempts == 20 - new RetryConfig([delay: '1s']).delay == Duration.of('1s') - new RetryConfig([maxDelay: '1m']).maxDelay == Duration.of('1m') - new RetryConfig([jitter: '0.5']).jitter == 0.5d - - } - - def 'should get the setting from the system env' () { - given: - SysEnv.push([ - NXF_RETRY_POLICY_DELAY: '10s', - NXF_RETRY_POLICY_MAX_DELAY: '100s', - NXF_RETRY_POLICY_MAX_ATTEMPTS: '1000', - NXF_RETRY_POLICY_JITTER: '10000' - ]) - - expect: - new RetryConfig().getDelay() == Duration.of('10s') - new RetryConfig().getMaxDelay() == Duration.of('100s') - new RetryConfig().getMaxAttempts() == 1000 - new RetryConfig().getJitter() == 10_000 - - cleanup: - SysEnv.pop() - } - -} diff --git a/modules/nf-commons/build.gradle b/modules/nf-commons/build.gradle index a54bf19a45..807536f2fe 100644 --- a/modules/nf-commons/build.gradle +++ b/modules/nf-commons/build.gradle @@ -34,6 +34,8 @@ dependencies { api 'org.pf4j:pf4j:3.12.0' api 'org.pf4j:pf4j-update:2.3.0' api 'dev.failsafe:failsafe:3.1.0' + api 'io.seqera:lib-httpx:1.0.0' + api 'io.seqera:lib-retry:1.2.0' // patch gson dependency required by pf4j api 'com.google.code.gson:gson:2.13.1' diff --git a/modules/nf-commons/src/main/nextflow/plugin/HttpPluginRepository.groovy b/modules/nf-commons/src/main/nextflow/plugin/HttpPluginRepository.groovy index f006382b36..4df12c6fa2 100644 --- a/modules/nf-commons/src/main/nextflow/plugin/HttpPluginRepository.groovy +++ b/modules/nf-commons/src/main/nextflow/plugin/HttpPluginRepository.groovy @@ -1,27 +1,20 @@ package nextflow.plugin + +import java.net.http.HttpRequest +import java.net.http.HttpResponse + import com.google.gson.Gson -import dev.failsafe.Failsafe -import dev.failsafe.FailsafeExecutor -import dev.failsafe.Fallback -import dev.failsafe.RetryPolicy -import dev.failsafe.event.EventListener -import dev.failsafe.event.ExecutionAttemptedEvent -import dev.failsafe.function.CheckedSupplier import groovy.transform.CompileStatic import groovy.util.logging.Slf4j +import io.seqera.http.HxClient import nextflow.BuildInfo +import nextflow.util.RetryConfig import org.pf4j.PluginRuntimeException import org.pf4j.update.FileDownloader import org.pf4j.update.FileVerifier import org.pf4j.update.PluginInfo -import org.pf4j.update.SimpleFileDownloader import org.pf4j.update.verifier.CompoundVerifier - -import java.net.http.HttpClient -import java.net.http.HttpRequest -import java.net.http.HttpResponse - /** * Represents an update repository served via an HTTP api. * @@ -35,11 +28,11 @@ import java.net.http.HttpResponse @Slf4j @CompileStatic class HttpPluginRepository implements PrefetchUpdateRepository { - private final HttpClient client = HttpClient.newHttpClient() private final String id private final URI url + private final HxClient httpClient - private Map plugins = new HashMap<>() + private Map plugins HttpPluginRepository(String id, URI url) { this.id = id @@ -47,6 +40,7 @@ class HttpPluginRepository implements PrefetchUpdateRepository { this.url = !url.toString().endsWith("/") ? URI.create(url.toString() + "/") : url + this.httpClient = HxClient.create(RetryConfig.config()) } // NOTE ON PREFETCHING @@ -83,7 +77,7 @@ class HttpPluginRepository implements PrefetchUpdateRepository { @Override Map getPlugins() { - if (plugins.isEmpty()) { + if (plugins==null) { log.warn "getPlugins() called before prefetch() - plugins map will be empty" return Map.of() } @@ -120,59 +114,53 @@ class HttpPluginRepository implements PrefetchUpdateRepository { private Map fetchMetadata(Collection specs) { final ordered = specs.sort(false) - final CheckedSupplier> supplier = () -> fetchMetadata0(ordered) - return retry().get(supplier) + return fetchMetadata0(ordered) } private Map fetchMetadata0(List specs) { - final gson = new Gson() - - def reqBody = gson.toJson([ - 'nextflowVersion': BuildInfo.version, - 'plugins' : specs - ]) - + def pluginsParam = specs.collect { "${it.id}${it.version ? '@' + it.version : ''}" }.join(',') + def uri = url.resolve("v1/plugins/dependencies?plugins=${URLEncoder.encode(pluginsParam, 'UTF-8')}&nextflowVersion=${URLEncoder.encode(BuildInfo.version, 'UTF-8')}") def req = HttpRequest.newBuilder() - .uri(url.resolve("plugins/collect")) - .POST(HttpRequest.BodyPublishers.ofString(reqBody)) + .uri(uri) + .GET() .build() - - def rep = client.send(req, HttpResponse.BodyHandlers.ofString()) - if (rep.statusCode() != 200) throw new PluginRuntimeException(errorMessage(rep, gson)) - try { - def repBody = gson.fromJson(rep.body(), FetchResponse) - return repBody.plugins.collectEntries { p -> Map.entry(p.id, p) } - } catch (Exception e) { - log.info("Plugin metadata response body: '${rep.body()}'") - throw new PluginRuntimeException("Failed to parse response body", e) + return sendAndParse(req) } - } - - // create a retry executor using failsafe - private static FailsafeExecutor retry() { - EventListener logAttempt = (ExecutionAttemptedEvent attempt) -> { - log.debug("Retrying download of plugins metadata - attempt ${attempt.attemptCount}, ${attempt.lastFailure.message}", attempt.lastFailure) + catch (PluginRuntimeException e) { + throw e } - Fallback fallback = Fallback.ofException { e -> - e.lastFailure instanceof ConnectException - ? new ConnectException("Failed to download plugins metadata") - : new PluginRuntimeException("Failed to download plugin metadata: ${e.lastFailure.message}") + catch (Exception e) { + throw new PluginRuntimeException(e, "Unable to connect to ${uri}- cause: ${e.message}") } - final policy = RetryPolicy.builder() - .withMaxAttempts(3) - .handle(ConnectException) - .onRetry(logAttempt) - .build() - return Failsafe.with(fallback, policy) } - private static String errorMessage(HttpResponse rep, Gson gson) { + private Map sendAndParse(HttpRequest req) { + final gson = new Gson() + final resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString()) + final body = resp.body() + log.debug "Registry request: ${resp.uri()}\n- code: ${resp.statusCode()}\n- body: ${body}" + if( resp.statusCode() != 200 ) { + final msg = "Invalid response while fetching plugin metadata from: ${req.uri()}\n- http status: ${resp.statusCode()}\n- response : ${body}" + throw new PluginRuntimeException(msg) + } try { - def err = gson.fromJson(rep.body(), ErrorResponse) - return "${err.type} - ${err.message}" - } catch (Exception e) { - return rep.body() + final FetchResponse decoded = gson.fromJson(body,FetchResponse) + if( decoded.plugins == null ) { + throw new PluginRuntimeException("Failed to download plugin metadata: Failed to parse response body") + } + final result = new HashMap() + for( PluginInfo plugin : decoded.plugins ) { + if( plugin.releases ) + result.put(plugin.id, plugin) + else + log.debug "Registry ${resp.uri().host} has no releases for plugin: ${plugin}" + } + return result + } + catch( Exception e ) { + final msg = "Unexpected error while fetching plugin metadata from: ${req.uri()}\n- message : ${e.message}\n- response: ${body}" + throw new PluginRuntimeException(msg) } } @@ -185,8 +173,4 @@ class HttpPluginRepository implements PrefetchUpdateRepository { List plugins } - private static class ErrorResponse { - String type - String message - } } diff --git a/modules/nf-commons/src/main/nextflow/plugin/OciAwareFileDownloader.groovy b/modules/nf-commons/src/main/nextflow/plugin/OciAwareFileDownloader.groovy index 1a6f0d0c06..8e8e10aa1d 100644 --- a/modules/nf-commons/src/main/nextflow/plugin/OciAwareFileDownloader.groovy +++ b/modules/nf-commons/src/main/nextflow/plugin/OciAwareFileDownloader.groovy @@ -17,14 +17,19 @@ package nextflow.plugin -import groovy.transform.CompileStatic -import groovy.util.logging.Slf4j -import org.pf4j.update.SimpleFileDownloader - +import java.net.http.HttpClient +import java.net.http.HttpRequest +import java.net.http.HttpResponse import java.nio.file.Files import java.nio.file.Path +import java.nio.file.StandardCopyOption +import java.time.Duration import java.util.regex.Pattern +import groovy.transform.CompileStatic +import groovy.util.logging.Slf4j +import nextflow.BuildInfo +import org.pf4j.update.SimpleFileDownloader /** * FileDownloader extension that enables the download of OCI compliant artifact that require a token authorization. * @@ -35,69 +40,172 @@ import java.util.regex.Pattern class OciAwareFileDownloader extends SimpleFileDownloader { private static final Pattern WWW_AUTH_PATTERN = ~/Bearer realm="([^"]+)",\s*service="([^"]+)",\s*scope="([^"]+)"/ + private static final Set REDIRECT_STATUS_CODES = [301, 302, 303, 307, 308] as Set + private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(90) /** * OCI aware download with token authorization. Tries to download the artifact and if it fails checks the headers to get the * @param fileUrl source file * @return */ + private final HttpClient httpClient = HttpClient.newBuilder() + .followRedirects(HttpClient.Redirect.NEVER) + .connectTimeout(Duration.ofSeconds(30)) + .build() + @Override protected Path downloadFileHttp(URL fileUrl) { + final destination = Files.createTempFile("nf-plugin", ".zip") - Path destination = Files.createTempDirectory("pf4j-update-downloader"); - destination.toFile().deleteOnExit(); + // sendRequest now handles redirects and authentication internally + HttpResponse response = sendRequest(fileUrl) - String path = fileUrl.getPath(); - String fileName = path.substring(path.lastIndexOf('/') + 1); - Path file = destination.resolve(fileName); - HttpURLConnection conn = (HttpURLConnection) fileUrl.openConnection() - conn.instanceFollowRedirects = true + // Check for HTTP error status codes + if (response.statusCode() >= 400) { + closeResponse(response) + throw new IOException("HTTP error ${response.statusCode()} downloading from $fileUrl") + } - if (conn.responseCode == HttpURLConnection.HTTP_UNAUTHORIZED) { - def wwwAuth = conn.getHeaderField("WWW-Authenticate") - if (wwwAuth?.contains("Bearer")) { - log.debug("Received 401 — attempting OCI token auth") + // Save the byte stream response directly to file + return downloadFileFromResponse(response, destination) + } - def matcher = WWW_AUTH_PATTERN.matcher(wwwAuth) - if (!matcher.find()) { - throw new IOException("Invalid WWW-Authenticate header: $wwwAuth") + private HttpRequest.Builder createRequestBuilder(URL url) { + return HttpRequest.newBuilder() + .uri(url.toURI()) + .header("User-Agent", "Nextflow/$BuildInfo.version") + .header("X-Nextflow-Version", BuildInfo.version) + .timeout(REQUEST_TIMEOUT) + .GET() + } + + private HttpResponse sendRequest0(URL url, String token = null) { + def requestBuilder = createRequestBuilder(url) + if (token) { + requestBuilder.header("Authorization", "Bearer $token") + } + HttpRequest request = requestBuilder.build() + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()) + log.debug "HTTP response from $url: status=${response.statusCode()}" + return response + } + + private HttpResponse sendRequest(URL url) { + Set attemptedUrls = new HashSet<>() + URL currentUrl = url + String token = null + + while (true) { + // submit the request + HttpResponse response = sendRequest0(currentUrl, token) + + // Handle redirects + if (response.statusCode() in REDIRECT_STATUS_CODES) { + // Prevent infinite redirect loops + if (attemptedUrls.contains(currentUrl.toString())) { + throw new IOException("Redirect loop detected for URL: $currentUrl") } - - def (realm, service, scope) = [matcher.group(1), matcher.group(2), matcher.group(3)] - def tokenUrl = "${realm}?service=${URLEncoder.encode(service, 'UTF-8')}&scope=${URLEncoder.encode(scope, 'UTF-8')}" - def token = fetchToken(tokenUrl) - - // Retry download with Bearer token - def authConn = (HttpURLConnection) fileUrl.openConnection() - authConn.setRequestProperty("Authorization", "Bearer $token") - authConn.instanceFollowRedirects = true - - authConn.inputStream.withStream { input -> - file.withOutputStream { out -> out << input } + attemptedUrls.add(currentUrl.toString()) + + String newUrl = response.headers().firstValue("Location").orElse(null) + if (newUrl) { + log.debug "Following redirect from $currentUrl to $newUrl" + currentUrl = URI.create(newUrl).toURL() + token = null + continue } - - return file } + // Handle authentication - retry once with token + if (response.statusCode() == 401 && token == null) { + token = handleAuthentication(response) + log.debug "Retrying request with authentication token" + continue + } + + // Check if response is successful before making InputStream request + if (response.statusCode() >= 400) { + throw new IOException("HTTP error ${response.statusCode()} from $currentUrl") + } + + // Now make final request with InputStream handler + def requestBuilder = createRequestBuilder(currentUrl) + if (token) { + requestBuilder.header("Authorization", "Bearer $token") + } + HttpRequest request = requestBuilder.build() + HttpResponse streamResponse = httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream()) + log.debug "HTTP stream response from $currentUrl: status=${streamResponse.statusCode()}" + return streamResponse } - - // Fallback to default behavior - conn.inputStream.withStream { input -> - file.withOutputStream { out -> out << input } + } + + private String handleAuthentication(HttpResponse response) { + log.debug("Received 401 - attempting OCI token auth") + def wwwAuth = response.headers().firstValue("WWW-Authenticate").orElse(null) + if (!wwwAuth?.contains("Bearer")) { + throw new IOException("Unsupported authentication method") + } + + def matcher = WWW_AUTH_PATTERN.matcher(wwwAuth) + if (!matcher.find()) { + throw new IOException("Invalid WWW-Authenticate header: $wwwAuth") + } + + def (realm, service, scope) = [matcher.group(1), matcher.group(2), matcher.group(3)] + def tokenUrl = "${realm}?service=${URLEncoder.encode(service, 'UTF-8')}&scope=${URLEncoder.encode(scope, 'UTF-8')}" + return fetchToken(tokenUrl) + } + + private Path downloadFileFromResponse(HttpResponse response, Path destination) { + log.debug "Saving downloaded file to: $destination" + + try (InputStream inputStream = response.body()) { + Files.copy(inputStream, destination, StandardCopyOption.REPLACE_EXISTING) + } + finally { + closeResponse(response) } - return file + return destination } + private String fetchToken(String tokenUrl) { - def conn = (HttpURLConnection) URI.create(tokenUrl).toURL().openConnection() - conn.setRequestProperty("Accept", "application/json") - - def json = conn.inputStream.getText("UTF-8") + HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create(tokenUrl)) + .header("Accept", "application/json") + .header("User-Agent", "Nextflow/$BuildInfo.version") + .header("X-Nextflow-Version", BuildInfo.version) + .timeout(REQUEST_TIMEOUT) + .GET() + .build() + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()) + log.debug "HTTP token response from $tokenUrl: status=${response.statusCode()}" + + if (response.statusCode() >= 400) { + throw new IOException("HTTP error ${response.statusCode()} fetching token from $tokenUrl") + } + + def json = response.body() def matcher = json =~ /"token"\s*:\s*"([^"]+)"/ if (matcher.find()) { return matcher.group(1) } throw new IOException("Token not found in response: $json") } + + static void closeResponse(HttpResponse response) { + log.trace "Closing HttpClient response: $response" + try { + // close the httpclient response to prevent leaks + // https://bugs.openjdk.org/browse/JDK-8308364 + final b0 = response.body() + if( b0 instanceof Closeable ) + b0.close() + } + catch (Throwable e) { + log.debug "Unexpected error while closing http response - cause: ${e.message}", e + } + } } diff --git a/modules/nf-commons/src/main/nextflow/plugin/PluginUpdater.groovy b/modules/nf-commons/src/main/nextflow/plugin/PluginUpdater.groovy index c4072592bc..62cfb590f5 100644 --- a/modules/nf-commons/src/main/nextflow/plugin/PluginUpdater.groovy +++ b/modules/nf-commons/src/main/nextflow/plugin/PluginUpdater.groovy @@ -239,10 +239,8 @@ class PluginUpdater extends UpdateManager { // 2. download to temporary location Path downloaded = safeDownloadPlugin(id, version); - // 3. rename if filename is sha digest - // when the plugin is downloaded from the (OCI) registry, it is named as "sha256:" - // rename it to something meaningful using the target plugin path - if ( downloaded.getFileName().toString().startsWith("sha256:")) { + // 3. rename to match the expected name + if ( downloaded.getFileName().toString() != "${pluginPath.getFileName()}.zip" ) { final targetName = downloaded.resolveSibling("${pluginPath.getFileName()}.zip") if ( !Files.move(downloaded, targetName) ) throw new PluginRuntimeException("Failed to rename '$downloaded'") downloaded = targetName diff --git a/modules/nf-commons/src/main/nextflow/plugin/PluginsFacade.groovy b/modules/nf-commons/src/main/nextflow/plugin/PluginsFacade.groovy index 18c331a073..fdef2c3e38 100644 --- a/modules/nf-commons/src/main/nextflow/plugin/PluginsFacade.groovy +++ b/modules/nf-commons/src/main/nextflow/plugin/PluginsFacade.groovy @@ -61,7 +61,7 @@ class PluginsFacade implements PluginStateListener { PluginsFacade() { mode = getPluginsMode() root = getPluginsDir() - indexUrl = getPluginsIndexUrl() + indexUrl = getPluginsRegistryUrl() offline = env.get('NXF_OFFLINE') == 'true' if( mode==DEV_MODE && root.toString()=='plugins' && !isRunningFromDistArchive() ) root = detectPluginsDevRoot() @@ -119,13 +119,13 @@ class PluginsFacade implements PluginStateListener { || hostname.endsWith('.seqera.io') } - protected String getPluginsIndexUrl() { - final url = env.get('NXF_PLUGINS_INDEX_URL') + protected String getPluginsRegistryUrl() { + final url = env.get('NXF_PLUGINS_REGISTRY_URL') if( !url ) { log.trace "Using default plugins url" return DEFAULT_PLUGINS_REPO } - log.debug "Detected NXF_PLUGINS_INDEX_URL=$url" + log.debug "Detected NXF_PLUGINS_REGISTRY_URL=$url" if( !isSupportedIndex(url) ) { // warn that this is experimental behaviour log.warn """\ diff --git a/modules/nf-commons/src/main/nextflow/util/RetryConfig.groovy b/modules/nf-commons/src/main/nextflow/util/RetryConfig.groovy new file mode 100644 index 0000000000..a73bae70c3 --- /dev/null +++ b/modules/nf-commons/src/main/nextflow/util/RetryConfig.groovy @@ -0,0 +1,148 @@ +/* + * Copyright 2013-2025, 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.util + +import com.google.common.base.CaseFormat +import groovy.transform.CompileDynamic +import groovy.transform.CompileStatic +import groovy.transform.EqualsAndHashCode +import groovy.transform.ToString +import groovy.util.logging.Slf4j +import io.seqera.util.retry.Retryable +import nextflow.Global +import nextflow.SysEnv +/** + * Models retry policy configuration + * + * @author Paolo Di Tommaso + */ +@Slf4j +@ToString(includePackage = false, includeNames = true) +@EqualsAndHashCode +@CompileStatic +class RetryConfig implements Retryable.Config { + + private final static Duration DEFAULT_DELAY = Duration.of('350ms') + private final static Duration DEFAULT_MAX_DELAY = Duration.of('90s') + private final static Integer DEFAULT_MAX_ATTEMPTS = 5 + private final static Double DEFAULT_JITTER = 0.25 + static final public double DEFAULT_MULTIPLIER = 2.0 + + private final static String ENV_PREFIX = 'NXF_RETRY_POLICY_' + + final private Duration delay + final private Duration maxDelay + final private int maxAttempts + final private double jitter + final private double multiplier + + RetryConfig() { + this(Collections.emptyMap()) + } + + RetryConfig(Map config) { + delay = + valueOf(config, 'delay', ENV_PREFIX, DEFAULT_DELAY, Duration) + maxDelay = + valueOf(config, 'maxDelay', ENV_PREFIX, DEFAULT_MAX_DELAY, Duration) + maxAttempts = + valueOf(config, 'maxAttempts', ENV_PREFIX, DEFAULT_MAX_ATTEMPTS, Integer) + jitter = + valueOf(config, 'jitter', ENV_PREFIX, DEFAULT_JITTER, Double) + multiplier = + valueOf(config, 'multiplier', ENV_PREFIX, DEFAULT_MULTIPLIER, Double) + } + + java.time.Duration getDelay() { + return java.time.Duration.ofMillis(delay.toMillis()) + } + + java.time.Duration getMaxDelay() { + return java.time.Duration.ofMillis(maxDelay.toMillis()) + } + + @Override + int getMaxAttempts() { + return maxAttempts } + + @Override + double getJitter() { + return jitter + } + + @Override + double getMultiplier() { + return multiplier + } + + static RetryConfig config() { + config(Global.config) + } + + static RetryConfig config(Map config) { + if( config!=null ) { + return new RetryConfig(getNestedConfig(config, 'nextflow', 'retryPolicy') ?: Collections.emptyMap()) + } + log.warn "Missing nextflow session - using default retry config" + return new RetryConfig() + } + + private static Map getNestedConfig(Map config, String... keys) { + def current = config + for (String key : keys) { + if (current instanceof Map && current.containsKey(key)) { + current = current.get(key) + } else { + return null + } + } + return current instanceof Map ? current : null + } + + static T valueOf(Map config, String name, String prefix, T defValue, Class type) { + assert name, "Argument 'name' cannot be null or empty" + assert type, "Argument 'type' cannot be null" + + // try to get the value from the config map + final cfg = config?.get(name) + if( cfg != null ) { + return toType(cfg, type) + } + // try to fallback to the sys environment + if( !prefix.endsWith('_') ) + prefix += '_' + final key = prefix.toUpperCase() + CaseFormat.LOWER_CAMEL.to(CaseFormat.UPPER_UNDERSCORE, name) + final env = SysEnv.get(key) + if( env != null ) { + return toType(env, type) + } + // return the default value + return defValue + } + + @CompileDynamic + static protected T toType(Object value, Class type) { + if( value == null ) + return null + if( type==Boolean.class ) { + return type.cast(Boolean.valueOf(value.toString())) + } + else { + return value.asType(type) + } + } +} diff --git a/modules/nf-commons/src/test/nextflow/plugin/HttpPluginRepositoryTest.groovy b/modules/nf-commons/src/test/nextflow/plugin/HttpPluginRepositoryTest.groovy index ac49fbe5af..7e8b9eaf54 100644 --- a/modules/nf-commons/src/test/nextflow/plugin/HttpPluginRepositoryTest.groovy +++ b/modules/nf-commons/src/test/nextflow/plugin/HttpPluginRepositoryTest.groovy @@ -1,15 +1,15 @@ package nextflow.plugin +import java.time.ZoneOffset +import java.time.ZonedDateTime + import com.github.tomakehurst.wiremock.junit.WireMockRule import com.github.tomjankes.wiremock.WireMockGroovy -import dev.failsafe.FailsafeException +import nextflow.BuildInfo import org.junit.Rule import org.pf4j.PluginRuntimeException import spock.lang.Specification -import java.time.ZoneOffset -import java.time.ZonedDateTime - class HttpPluginRepositoryTest extends Specification { @Rule WireMockRule wiremock = new WireMockRule(0) @@ -28,16 +28,15 @@ class HttpPluginRepositoryTest extends Specification { given: wm.stub { request { - method 'POST' - url '/plugins/collect' + method 'GET' + url "/v1/plugins/dependencies?plugins=nf-fake&nextflowVersion=${BuildInfo.version}" } response { status 200 body """{ "plugins": [ { - "id": "nf-fake", - "releases": [] + "id": "nf-fake" } ] } @@ -50,10 +49,7 @@ class HttpPluginRepositoryTest extends Specification { then: def plugins = unit.getPlugins() - plugins.size() == 1 - def p1 = plugins.get("nf-fake") - p1.id == "nf-fake" - p1.releases.size() == 0 + plugins.size() == 0 } // ------------------------------------------------------------------------ @@ -62,8 +58,8 @@ class HttpPluginRepositoryTest extends Specification { given: wm.stub { request { - method 'POST' - url '/plugins/collect' + method 'GET' + url "/v1/plugins/dependencies?plugins=nf-fake&nextflowVersion=${BuildInfo.version}" } response { status 200 @@ -114,8 +110,8 @@ class HttpPluginRepositoryTest extends Specification { unit.prefetch([new PluginSpec("nf-fake")]) then: - def err = thrown FailsafeException - err.message == "java.net.ConnectException: Failed to download plugins metadata" + def err = thrown PluginRuntimeException + err.message.startsWith("Unable to connect to http://localhost") } // ------------------------------------------------------------------------ @@ -124,8 +120,8 @@ class HttpPluginRepositoryTest extends Specification { given: wm.stub { request { - method 'POST' - url '/plugins/collect' + method 'GET' + url "/v1/plugins/dependencies?plugins=nf-fake&nextflowVersion=${BuildInfo.version}" } response { status 500 @@ -138,7 +134,10 @@ class HttpPluginRepositoryTest extends Specification { then: def err = thrown PluginRuntimeException - err.message == "Failed to download plugin metadata: Server error!" + err.message == """\ + Invalid response while fetching plugin metadata from: http://localhost:${wiremock.port()}/v1/plugins/dependencies?plugins=nf-fake&nextflowVersion=${BuildInfo.version} + - http status: 500 + - response : Server error!""".stripIndent() } // ------------------------------------------------------------------------ @@ -147,8 +146,8 @@ class HttpPluginRepositoryTest extends Specification { given: wm.stub { request { - method: 'POST' - url: '/plugins/collect' + method 'GET' + url "/v1/plugins/dependencies?plugins=nf-fake&nextflowVersion=${BuildInfo.version}" } response { status 200 @@ -168,7 +167,7 @@ class HttpPluginRepositoryTest extends Specification { then: def err = thrown PluginRuntimeException - err.message == "Failed to download plugin metadata: Failed to parse response body" + err.message.startsWith("Unexpected error while fetching plugin metadata from: http://localhost") } // ------------------------------------------------------------------------ @@ -177,8 +176,8 @@ class HttpPluginRepositoryTest extends Specification { given: wm.stub { request { - method: 'POST' - url: '/plugins/collect' + method 'GET' + url "/v1/plugins/dependencies?plugins=nf-fake&nextflowVersion=${BuildInfo.version}" } response { status 400 @@ -194,7 +193,7 @@ class HttpPluginRepositoryTest extends Specification { then: def err = thrown PluginRuntimeException - err.message == "Failed to download plugin metadata: SOME_ERROR - Unparseable request" + err.message.startsWith("Invalid response while fetching plugin metadata from: http://localhost") } // ------------------------------------------------------------------------ diff --git a/modules/nf-commons/src/test/nextflow/plugin/OciAwareFileDownloaderTest.groovy b/modules/nf-commons/src/test/nextflow/plugin/OciAwareFileDownloaderTest.groovy index cb2e7e1725..91cf79cdbd 100644 --- a/modules/nf-commons/src/test/nextflow/plugin/OciAwareFileDownloaderTest.groovy +++ b/modules/nf-commons/src/test/nextflow/plugin/OciAwareFileDownloaderTest.groovy @@ -17,6 +17,7 @@ package nextflow.plugin import com.github.tomakehurst.wiremock.junit.WireMockRule +import nextflow.BuildInfo import org.junit.Rule import spock.lang.Specification @@ -49,7 +50,7 @@ class OciAwareFileDownloaderTest extends Specification { downloadedFile != null Files.exists(downloadedFile) downloadedFile.text == fileContent - downloadedFile.fileName.toString() == "plugin.zip" + downloadedFile.fileName.toString().endsWith(".zip") cleanup: if (downloadedFile) Files.deleteIfExists(downloadedFile) @@ -126,4 +127,229 @@ class OciAwareFileDownloaderTest extends Specification { def ex = thrown(IOException) ex.message.contains("Token not found in response") } + + def 'should include User-Agent header in all HTTP requests'() { + given: + def expectedUserAgent = "Nextflow/$BuildInfo.version" + def fileContent = "test plugin content" + + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .withHeader("User-Agent", equalTo(expectedUserAgent)) + .willReturn(ok().withBody(fileContent)) + ) + + when: + def downloadedFile = downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + downloadedFile != null + Files.exists(downloadedFile) + downloadedFile.text == fileContent + + // Verify the request was made with correct User-Agent + wiremock.verify(getRequestedFor(urlPathEqualTo("/plugin.zip")) + .withHeader("User-Agent", equalTo(expectedUserAgent)) + ) + + cleanup: + Files.deleteIfExists(downloadedFile) + } + + def 'should include X-Nextflow-Version header in all HTTP requests'() { + given: + def expectedVersion = BuildInfo.version + def fileContent = "test plugin content" + + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .withHeader("X-Nextflow-Version", equalTo(expectedVersion)) + .willReturn(ok().withBody(fileContent)) + ) + + when: + def downloadedFile = downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + downloadedFile != null + Files.exists(downloadedFile) + downloadedFile.text == fileContent + + // Verify the request was made with correct X-Nextflow-Version + wiremock.verify(getRequestedFor(urlPathEqualTo("/plugin.zip")) + .withHeader("X-Nextflow-Version", equalTo(expectedVersion)) + ) + + cleanup: + Files.deleteIfExists(downloadedFile) + } + + def 'should include both User-Agent and X-Nextflow-Version headers in token requests'() { + given: + def expectedUserAgent = "Nextflow/$BuildInfo.version" + def expectedVersion = BuildInfo.version + def fileContent = "authenticated plugin archive content" + def tokenResponse = '{"token": "test-bearer-token-12345"}' + def authServerUrl = "${wiremock.baseUrl()}/token" + def wwwAuthHeader = "Bearer realm=\"${authServerUrl}\",service=\"registry.example.com\",scope=\"repository:plugins/nf-test:pull\"" + + // Token endpoint expects both headers + wiremock.stubFor(get(urlPathEqualTo("/token")) + .withQueryParam('service', equalTo('registry.example.com')) + .withQueryParam('scope', equalTo('repository:plugins/nf-test:pull')) + .withHeader("User-Agent", equalTo(expectedUserAgent)) + .withHeader("X-Nextflow-Version", equalTo(expectedVersion)) + .willReturn(okJson(tokenResponse)) + ) + + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .willReturn(unauthorized().withHeader("WWW-Authenticate", wwwAuthHeader)) + ) + + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .withHeader("Authorization", equalTo("Bearer test-bearer-token-12345")) + .withHeader("User-Agent", equalTo(expectedUserAgent)) + .withHeader("X-Nextflow-Version", equalTo(expectedVersion)) + .willReturn(ok().withBody(fileContent)) + ) + + when: + def downloadedFile = downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + downloadedFile != null + Files.exists(downloadedFile) + downloadedFile.text == fileContent + + // Verify token request had both headers + wiremock.verify(getRequestedFor(urlPathEqualTo("/token")) + .withHeader("User-Agent", equalTo(expectedUserAgent)) + .withHeader("X-Nextflow-Version", equalTo(expectedVersion)) + ) + + // Verify authenticated download request had both headers + wiremock.verify(getRequestedFor(urlPathEqualTo("/plugin.zip")) + .withHeader("Authorization", equalTo("Bearer test-bearer-token-12345")) + .withHeader("User-Agent", equalTo(expectedUserAgent)) + .withHeader("X-Nextflow-Version", equalTo(expectedVersion)) + ) + + cleanup: + Files.deleteIfExists(downloadedFile) + } + + def 'should handle redirects automatically'() { + given: + def fileContent = "redirected plugin content" + def redirectUrl = "${wiremock.baseUrl()}/redirected/plugin.zip" + + // Initial request returns redirect + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .willReturn(temporaryRedirect(redirectUrl)) + ) + + // Redirected URL returns the file + wiremock.stubFor(get(urlPathEqualTo("/redirected/plugin.zip")) + .willReturn(ok().withBody(fileContent)) + ) + + when: + def downloadedFile = downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + downloadedFile != null + Files.exists(downloadedFile) + downloadedFile.text == fileContent + + // Verify both requests were made + wiremock.verify(getRequestedFor(urlPathEqualTo("/plugin.zip"))) + wiremock.verify(getRequestedFor(urlPathEqualTo("/redirected/plugin.zip"))) + + cleanup: + Files.deleteIfExists(downloadedFile) + } + + def 'should detect and prevent redirect loops'() { + given: + def redirectUrl1 = "${wiremock.baseUrl()}/redirect1" + def redirectUrl2 = "${wiremock.baseUrl()}/redirect2" + + // Create a redirect loop + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .willReturn(temporaryRedirect(redirectUrl1)) + ) + wiremock.stubFor(get(urlPathEqualTo("/redirect1")) + .willReturn(temporaryRedirect(redirectUrl2)) + ) + wiremock.stubFor(get(urlPathEqualTo("/redirect2")) + .willReturn(temporaryRedirect("${wiremock.baseUrl()}/plugin.zip")) + ) + + when: + downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + def ex = thrown(IOException) + ex.message.contains("Redirect loop detected") + } + + def 'should handle authentication after redirects'() { + given: + def fileContent = "authenticated after redirect content" + def tokenResponse = '{"token": "redirect-auth-token"}' + def authServerUrl = "${wiremock.baseUrl()}/token" + def wwwAuthHeader = "Bearer realm=\"${authServerUrl}\",service=\"registry.example.com\",scope=\"repository:plugins/nf-test:pull\"" + def redirectUrl = "${wiremock.baseUrl()}/auth/plugin.zip" + + // Initial request redirects + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .willReturn(temporaryRedirect(redirectUrl)) + ) + + // Redirected URL requires authentication + wiremock.stubFor(get(urlPathEqualTo("/auth/plugin.zip")) + .willReturn(unauthorized().withHeader("WWW-Authenticate", wwwAuthHeader)) + ) + + // Token endpoint + wiremock.stubFor(get(urlPathEqualTo("/token")) + .withQueryParam('service', equalTo('registry.example.com')) + .withQueryParam('scope', equalTo('repository:plugins/nf-test:pull')) + .willReturn(okJson(tokenResponse)) + ) + + // Authenticated request succeeds + wiremock.stubFor(get(urlPathEqualTo("/auth/plugin.zip")) + .withHeader("Authorization", equalTo("Bearer redirect-auth-token")) + .willReturn(ok().withBody(fileContent)) + ) + + when: + def downloadedFile = downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + downloadedFile != null + Files.exists(downloadedFile) + downloadedFile.text == fileContent + + // Verify the sequence: initial request, redirect, auth challenge, token request, authenticated download + wiremock.verify(getRequestedFor(urlPathEqualTo("/plugin.zip"))) + wiremock.verify(3, getRequestedFor(urlPathEqualTo("/auth/plugin.zip"))) // String request for auth check, 2 successful requests (String + InputStream) + wiremock.verify(getRequestedFor(urlPathEqualTo("/token"))) + + cleanup: + Files.deleteIfExists(downloadedFile) + } + + def 'should throw IOException for HTTP error status codes'() { + given: + wiremock.stubFor(get(urlPathEqualTo("/plugin.zip")) + .willReturn(serverError()) + ) + + when: + downloader.downloadFileHttp(new URL("${wiremock.baseUrl()}/plugin.zip")) + + then: + def ex = thrown(IOException) + ex.message.contains("HTTP error 500") + } } \ No newline at end of file diff --git a/modules/nf-commons/src/test/nextflow/util/RetryConfigTest.groovy b/modules/nf-commons/src/test/nextflow/util/RetryConfigTest.groovy new file mode 100644 index 0000000000..2432433d6c --- /dev/null +++ b/modules/nf-commons/src/test/nextflow/util/RetryConfigTest.groovy @@ -0,0 +1,133 @@ +/* + * Copyright 2013-2025, 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.util + +import nextflow.SysEnv +import spock.lang.Specification +import spock.lang.Unroll + +/** + * + * @author Ben Sherman + */ +class RetryConfigTest extends Specification { + + def 'should create retry config' () { + expect: + new RetryConfig().delay == java.time.Duration.ofMillis(350) + new RetryConfig().maxDelay == java.time.Duration.ofSeconds(90) + new RetryConfig().maxAttempts == 5 + new RetryConfig().jitter == 0.25d + new RetryConfig().multiplier == 2d + + and: + new RetryConfig([maxAttempts: 20]).maxAttempts == 20 + new RetryConfig([delay: '1s']).delay == java.time.Duration.ofSeconds(1) + new RetryConfig([maxDelay: '1m']).maxDelay == java.time.Duration.ofMinutes(1) + new RetryConfig([jitter: '0.5']).jitter == 0.5d + new RetryConfig([multiplier: '5']).multiplier == 5d + } + + def 'should get the setting from the system env' () { + given: + SysEnv.push([ + NXF_RETRY_POLICY_DELAY: '10s', + NXF_RETRY_POLICY_MAX_DELAY: '100s', + NXF_RETRY_POLICY_MAX_ATTEMPTS: '1000', + NXF_RETRY_POLICY_JITTER: '10000', + NXF_RETRY_POLICY_MULTIPLIER: '90' + ]) + + expect: + new RetryConfig().getDelay() == java.time.Duration.ofSeconds(10) + new RetryConfig().getMaxDelay() == java.time.Duration.ofSeconds(100) + new RetryConfig().getMaxAttempts() == 1000 + new RetryConfig().getJitter() == 10_000 + new RetryConfig().getMultiplier() == 90d + + cleanup: + SysEnv.pop() + } + + @Unroll + def 'should get config from map' () { + given: + def NAME = 'foo' + def PREFIX = 'P_' + and: + SysEnv.push(ENV) + + expect: + RetryConfig.valueOf(CONFIG, NAME, PREFIX, DEF_VAL, DEF_TYPE) == EXPECTED + + cleanup: + SysEnv.pop() + + where: + CONFIG | ENV | DEF_VAL | DEF_TYPE | EXPECTED + null | [:] | null | String | null + [:] | [:] | null | String | null + [:] | [:] | 'one' | String | 'one' + [foo:'two'] | [:] | 'one' | String | 'two' + [foo:''] | [:] | 'one' | String | '' + [foo:'two'] | [P_FOO:'bar'] | 'one' | String | 'two' + [:] | [P_FOO:'bar'] | 'one' | String | 'bar' + + and: + null | [:] | null | Integer | null + [:] | [:] | null | Integer | null + [:] | [:] | 1 | Integer | 1 + [foo:2] | [:] | 1 | Integer | 2 + [foo:'2'] | [:] | 1 | Integer | 2 + [foo:'2'] | [P_FOO:'3'] | 1 | Integer | 2 + [:] | [P_FOO:'3'] | 1 | Integer | 3 + + and: + null | [:] | null | Boolean | null + [:] | [:] | true | Boolean | true + [foo:false] | [:] | true | Boolean | false + [foo:'false'] | [:] | true | Boolean | false + [foo:true] | [:] | false | Boolean | true + [foo:'true'] | [:] | false | Boolean | true + [foo:'true'] | [P_FOO:'false']| null | Boolean | true + [:] | [P_FOO:'false']| null | Boolean | false + [:] | [P_FOO:'true'] | null | Boolean | true + + and: + [:] | [:] | Duration.of('1s') | Duration | Duration.of('1s') + [foo:'10ms'] | [:] | null | Duration | Duration.of('10ms') + [:] | [P_FOO:'1s'] | null | Duration | Duration.of('1s') + } + + def 'should map camelCase to snake uppercase' () { + given: + SysEnv.push(ENV) + + expect: + RetryConfig.valueOf([:], NAME, PREFIX, null, String) == EXPECTED + + cleanup: + SysEnv.pop() + + where: + EXPECTED | PREFIX | NAME | ENV + null | 'foo' | 'bar' | [:] + 'one' | 'foo' | 'bar' | [FOO_BAR: 'one'] + 'one' | 'foo_' | 'bar' | [FOO_BAR: 'one'] + 'one' | 'foo_' | 'thisAndThat' | [FOO_THIS_AND_THAT: 'one'] + } +} diff --git a/tests/checks/plugin-registry.nf/.checks b/tests/checks/plugin-registry.nf/.checks index 517ee1719b..9a4746471e 100644 --- a/tests/checks/plugin-registry.nf/.checks +++ b/tests/checks/plugin-registry.nf/.checks @@ -4,7 +4,7 @@ set -e # run normal mode # echo '' -NXF_PLUGINS_INDEX_URL=https://plugin-registry.dev-tower.net/api NXF_PLUGINS_DIR=$PWD/.nextflow/plugins/ $NXF_RUN -plugins nf-ci-integration-test@0.1.0 | tee stdout +NXF_PLUGINS_REGISTRY_URL=https://plugin-registry.dev-tower.net/api NXF_PLUGINS_DIR=$PWD/.nextflow/plugins/ $NXF_RUN -plugins nf-ci-integration-test@0.1.0 | tee stdout [[ `grep 'INFO' .nextflow.log | grep -c 'Downloading plugin nf-ci-integration-test@0.1.0'` == 1 ]] || false [[ `grep -c 'Pipeline is starting using nf-ci-test-integration plugin' stdout` == 1 ]] || false