Fix Fusion license JWT token refresh by reusing TowerClient

This change addresses an issue where the Fusion license validation would fail after the Tower access token was refreshed. Previously, TowerFusionToken created a separate HTTP client with its own token manager, which meant it didn't receive updated refresh tokens when the main TowerClient refreshed the access token. Since OAuth refresh tokens are single-use, the old token held by the Fusion client would become invalid, causing authentication failures.

The fix modifies TowerFusionToken to reuse the shared TowerClient instance instead of creating its own HTTP client. This ensures that both workflow reporting and Fusion license validation share the same token manager and stay synchronized when tokens are refreshed.

Author: jorgee

plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerFusionToken.groovy

@@ -1,25 +1,19 @@
 package io.seqera.tower.plugin
 
-import java.net.http.HttpClient
-import java.net.http.HttpRequest
 import java.time.Duration
 import java.time.Instant
 import java.time.temporal.ChronoUnit
-import java.util.concurrent.Executors
 
 import com.google.common.cache.Cache
 import com.google.common.cache.CacheBuilder
 import com.google.common.util.concurrent.UncheckedExecutionException
 import com.google.gson.JsonSyntaxException
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
-import io.seqera.http.HxClient
-import io.seqera.http.HxConfig
 import io.seqera.tower.plugin.exception.BadResponseException
 import io.seqera.tower.plugin.exception.UnauthorizedException
 import io.seqera.tower.plugin.exchange.GetLicenseTokenRequest
 import io.seqera.tower.plugin.exchange.GetLicenseTokenResponse
-import io.seqera.util.trace.TraceUtils
 import nextflow.SysEnv
 import nextflow.exception.AbortOperationException
 import nextflow.exception.ReportWarningException
@@ -28,8 +22,6 @@ import nextflow.fusion.FusionToken
 import nextflow.platform.PlatformHelper
 import nextflow.plugin.Priority
 import nextflow.serde.gson.GsonEncoder
-import nextflow.util.RetryConfig
-import nextflow.util.Threads
 import org.pf4j.Extension
 /**
  * Environment provider for Platform-specific environment variables.
@@ -45,14 +37,8 @@ class TowerFusionToken implements FusionToken {
     // The path relative to the Platform endpoint where license-scoped JWT tokens are obtained
     private static final String LICENSE_TOKEN_PATH = 'license/token/'
 
-    // Server errors that should trigger a retry
-    private static final Set<Integer> SERVER_ERRORS = Set.of(408, 429, 500, 502, 503, 504)
-
-    // Default connection timeout for HTTP requests
-    private static final Duration DEFAULT_CONNECTION_TIMEOUT = Duration.of(30, ChronoUnit.SECONDS)
-
-    // The HttpClient instance used to send requests
-    private HxClient httpClient
+    // The TowerClient instance used to send requests
+    private TowerClient client
 
     // Time-to-live for cached tokens
     private Duration tokenTTL = Duration.of(1, ChronoUnit.HOURS)
@@ -84,7 +70,7 @@ class TowerFusionToken implements FusionToken {
         this.refreshToken = PlatformHelper.getRefreshToken(config, env)
         this.workflowId = env.get('TOWER_WORKFLOW_ID')
         this.workspaceId = PlatformHelper.getWorkspaceId(config, env)
-        this.httpClient = newDefaultHttpClient(accessToken, refreshToken)
+        this.client = TowerFactory.client()
     }
 
     protected void validateConfig() {
@@ -168,51 +154,6 @@ class TowerFusionToken implements FusionToken {
      * Helper methods
      *************************************************************************/
 
-    /**
-     * Create a new HttpClient instance with default settings
-     * @return The new HttpClient instance
-     */
-    private HxClient newDefaultHttpClient(String accessToken, String refreshToken) {
-        final refreshUrl = refreshToken ? "$endpoint/oauth/access_token" : null
-        // the client config
-        final config = HxConfig.newBuilder()
-            .bearerToken(accessToken)
-            .refreshToken(refreshToken)
-            .refreshTokenUrl(refreshUrl)
-            .refreshCookiePolicy(CookiePolicy.ACCEPT_ALL)
-            .retryStatusCodes(SERVER_ERRORS)
-            .retryConfig(RetryConfig.config())
-            .build()
-        // the client builder
-        final builder = HxClient.newBuilder()
-            .version(HttpClient.Version.HTTP_1_1)
-            .connectTimeout(DEFAULT_CONNECTION_TIMEOUT)
-            .config(config)
-        // use virtual threads executor if enabled
-        if ( Threads.useVirtual() ) {
-            builder.executor(Executors.newVirtualThreadPerTaskExecutor())
-        }
-        // build and return the new client
-        return builder.build()
-    }
-
-    /**
-     * Create a {@link HttpRequest} representing a {@link GetLicenseTokenRequest} object
-     *
-     * @param req The LicenseTokenRequest object
-     * @return The resulting HttpRequest object
-     */
-    private HttpRequest makeHttpRequest(GetLicenseTokenRequest req) {
-        final gson = new GsonEncoder<GetLicenseTokenRequest>() {}
-        final body = HttpRequest.BodyPublishers.ofString( gson.encode(req) )
-        return HttpRequest.newBuilder()
-            .uri(URI.create("${endpoint}/${LICENSE_TOKEN_PATH}").normalize())
-            .header('Content-Type', 'application/json')
-            .header('Traceparent', TraceUtils.rndTrace())
-            .POST(body)
-            .build()
-    }
-
     /**
      * Parse a JSON string into a {@link GetLicenseTokenResponse} object
      *
@@ -233,28 +174,20 @@ class TowerFusionToken implements FusionToken {
      * @return The LicenseTokenResponse object
      */
     private GetLicenseTokenResponse sendRequest(GetLicenseTokenRequest request) {
-        final httpReq = makeHttpRequest(request)
-        try {
-            final resp = httpClient.sendAsString(httpReq)
-            if( log.isTraceEnabled() || resp.statusCode()!=200 )
-                log.debug "Fusion license request ${httpReq} ${request}; status=${resp.statusCode()}; body: ${resp.body()}"
-            else 
-                log.debug "Fusion license request ${httpReq}; status=${resp.statusCode()}"
-            // check ok response
-            if( resp.statusCode() == 200 ) {
-                final ret = parseLicenseTokenResponse(resp.body())
-                return ret
-            }
-            // check for unauthorized error
-            if( resp.statusCode() == 401 ) {
-                throw new UnauthorizedException("Unauthorized [401] - Verify you have provided a Seqera Platform valid access token")
-            }
-            // unpexted error
-            throw new BadResponseException("Invalid response: ${httpReq.method()} ${httpReq.uri()} [${resp.statusCode()}] ${resp.body()}")
+        final url = "${client.getEndpoint()}/${LICENSE_TOKEN_PATH}"
+        final resp = client.sendHttpMessage(url, request.toMap())
+
+        if( resp.code == 200 ) {
+            final ret = parseLicenseTokenResponse(resp.message)
+            return ret
         }
-        catch (IOException e) {
-            throw new IllegalStateException("Unable to send request to '${httpReq.uri()}' : ${e.message}")
+
+        if( resp.code == 401 ) {
+            throw new UnauthorizedException("Unauthorized [401] - Verify you have provided a Seqera Platform valid access token")
         }
+
+        throw new BadResponseException("Invalid response: ${url} [${resp.code}] ${resp.message} -- ${resp.cause}")
+
     }
 
 }

plugins/nf-tower/src/main/io/seqera/tower/plugin/exchange/GetLicenseTokenRequest.groovy

@@ -29,4 +29,16 @@ class GetLicenseTokenRequest {
      * The Platform workspace ID associated with this request
      */
     String workspaceId
+
+    /**
+     * @return a Map representation of the request
+     */
+    Map<String, String> toMap() {
+        final map = new HashMap<String, String>()
+        map.product = this.product
+        map.version = this.version
+        map.workflowId = this.workflowId
+        map.workspaceId = this.workspaceId
+        return map
+    }
 }