Merge branch 'master' into nf-356-tower-abort-on-error

Author: jorgee

adr/20260310-seqera-dataset-filesystem.md

@@ -0,0 +1,136 @@
+# NIO Filesystem for Seqera Platform Datasets
+
+- Authors: Jorge Ejarque
+- Status: draft
+- Date: 2026-03-10
+- Tags: nio, filesystem, seqera, datasets, nf-tower
+
+Technical Story: Enable Nextflow pipelines to read Seqera Platform datasets as ordinary file paths using `seqera://` URIs.
+
+## Summary
+
+Add a Java NIO `FileSystemProvider` to the `nf-tower` plugin that registers the `seqera://` scheme, allowing pipelines to reference Seqera Platform datasets (CSV/TSV) as standard file paths without manual download steps. The implementation reuses the existing `TowerClient` for all HTTP communication, inheriting authentication and retry behaviour.
+
+## Problem Statement
+
+Nextflow users managing datasets on the Seqera Platform must currently download dataset files manually or through custom scripts before referencing them in pipelines. There is no native integration between Nextflow's file abstraction and the Seqera Platform dataset API. This creates friction in workflows where datasets are the primary input and forces users to handle authentication, versioning, and file staging outside the pipeline definition.
+
+## Goals or Decision Drivers
+
+- Transparent access to Seqera Platform datasets using standard Nextflow file path syntax
+- Reuse of existing nf-tower plugin infrastructure (authentication, HTTP client, retry/backoff)
+- Hierarchical path browsing matching the platform's org/workspace/dataset structure
+- Extensible architecture that can support future Seqera-managed resource types (e.g. data-links)
+- No new plugin or module — feature lives within nf-tower
+
+## Non-goals
+
+- Streaming large datasets — the Platform API does not support streaming; content is fully buffered on download
+- Implementing resource types beyond `datasets` — only the extensible architecture is required
+- Local caching across pipeline runs — Nextflow's standard task staging handles caching
+- Dataset management operations (delete, rename) — the filesystem is read-only in the initial implementation
+
+## Considered Options
+
+### Option 1: Standalone plugin with dedicated HTTP client
+
+A new `nf-seqera-fs` plugin with its own HTTP client configuration and authentication setup.
+
+- Good, because it isolates the filesystem code from the nf-tower plugin
+- Bad, because it duplicates authentication configuration and HTTP client setup
+- Bad, because two separate HTTP clients sharing a refresh token would corrupt each other's auth state
+
+### Option 2: NIO filesystem within nf-tower using TowerClient delegation
+
+Add the filesystem to nf-tower, delegating all HTTP through the existing `TowerClient` singleton via a typed `SeqeraDatasetClient` wrapper.
+
+- Good, because it shares authentication and token refresh with TowerClient
+- Good, because it reuses existing retry/backoff configuration
+- Good, because no new dependencies are needed
+
+### Option 3: Direct HxClient usage within nf-tower
+
+Add the filesystem to nf-tower but use `HxClient` directly rather than going through TowerClient.
+
+- Good, because it gives full control over request construction
+- Bad, because exposing HxClient internals couples the filesystem to implementation details
+- Bad, because token refresh coordination with TowerClient becomes manual
+
+## Solution or decision outcome
+
+Option 2 — NIO filesystem within nf-tower using TowerClient delegation. All HTTP calls go through `TowerClient.sendApiRequest()`, ensuring a single point of authentication and retry logic.
+
+## Rationale & discussion
+
+### Path Hierarchy
+
+The `seqera://` path encodes the Platform's organizational structure directly:
+
+```
+seqera://                                        → ROOT (directory, depth 0)
+  └── <org>/                                     → ORGANIZATION (directory, depth 1)
+        └── <workspace>/                         → WORKSPACE (directory, depth 2)
+              └── datasets/                      → RESOURCE TYPE (directory, depth 3)
+                    └── <name>[@<version>]        → DATASET (file, depth 4)
+```
+
+Each level is a directory except the leaf dataset, which is a file. Version pinning uses an `@version` suffix on the dataset name segment (e.g. `seqera://acme/research/datasets/samples@2`). Without it, the latest non-disabled version is resolved.
+
+### Name-to-ID Resolution
+
+The path uses human-readable names but the Platform API requires numeric IDs. Resolution is built from two API calls at filesystem initialization:
+
+1. `GET /user-info` → obtain `userId`
+2. `GET /user/{userId}/workspaces` → returns all accessible org/workspace pairs
+
+This single source provides both directory listing content and name→ID mapping. Results are cached in `SeqeraFileSystem` with invalidation on write operations. `GET /orgs` is intentionally not used as it returns all platform orgs, not scoped to user membership.
+
+### Component Structure
+
+```
+plugins/nf-tower/src/main/io/seqera/tower/plugin/
+├── fs/                             ← NIO layer
+│   ├── SeqeraFileSystemProvider    ← FileSystemProvider (scheme: "seqera")
+│   ├── SeqeraFileSystem            ← FileSystem with org/workspace/dataset caches
+│   ├── SeqeraPath                  ← Path implementation (depth 0–4)
+│   ├── SeqeraFileAttributes        ← BasicFileAttributes
+│   ├── SeqeraPathFactory           ← PF4J FileSystemPathFactory extension
+│   └── DatasetInputStream          ← SeekableByteChannel over InputStream
+├── dataset/                        ← API client layer
+│   ├── SeqeraDatasetClient         ← Typed HTTP client wrapping TowerClient
+│   ├── DatasetDto                  ← Dataset API response model
+│   ├── DatasetVersionDto           ← Version API response model
+│   ├── OrgAndWorkspaceDto          ← Org/workspace list model
+│   └── WorkspaceOrgDto             ← Workspace/org mapping model
+└── resources/META-INF/services/
+    └── java.nio.file.spi.FileSystemProvider
+```
+
+### Key Design Decisions
+
+1. **TowerClient delegation**: `SeqeraDatasetClient` delegates all HTTP through `TowerFactory.client()` → `TowerClient.sendApiRequest()`. This ensures shared authentication state and avoids the token refresh corruption that would occur with separate HTTP client instances.
+
+2. **One filesystem per JVM**: `SeqeraFileSystemProvider` maintains a single `SeqeraFileSystem` keyed by scheme. This matches the `TowerClient` singleton-per-session pattern.
+
+3. **Read-only initial scope**: The filesystem reports `isReadOnly()=true`. Write support (dataset upload via multipart POST) is deferred to a future iteration.
+
+4. **Download filename constraint**: The Platform API's download endpoint (`GET /datasets/{id}/v/{version}/n/{fileName}`) requires the exact filename from upload time. The implementation always resolves `DatasetVersionDto.fileName` from `GET /datasets/{id}/versions` before constructing the download URL.
+
+5. **Extensible resource types**: The path hierarchy reserves depth 3 for a resource type segment (currently only `datasets`). Adding support for data-links or other resource types requires only a new handler at the directory listing and I/O layers, with no changes to path resolution or authentication.
+
+6. **Thread safety**: `SeqeraFileSystem` cache methods and `SeqeraFileSystemProvider` lifecycle methods are `synchronized`. The filesystem map uses `LinkedHashMap` with external synchronization rather than `ConcurrentHashMap`, matching the low-contention access pattern.
+
+### Limitations
+
+- **No size metadata**: `SeqeraFileAttributes.size()` returns 0 for all paths because the Platform API does not expose content length in dataset metadata.
+- **Single endpoint per JVM**: The filesystem key is scheme-only; concurrent access to different Platform endpoints in the same JVM is not supported.
+
+### Streaming Downloads
+
+Dataset downloads use `TowerClient.sendStreamingRequest()` which calls `HxClient.sendAsStream()` — the response body is returned as an `InputStream` streamed directly from the HTTP connection. This avoids the triple-buffering problem (`String` → `getBytes()` → `ByteArrayInputStream`) that would otherwise consume ~40 MB heap per 10 MB dataset. The `HxClient.sendAsStream()` method goes through the same `sendWithRetry()` path as `sendAsString()`, so retry logic and token refresh are preserved.
+
+## Links
+
+- [Spec](../specs/260310-seqera-dataset-fs/spec.md)
+- [Implementation plan](../specs/260310-seqera-dataset-fs/plan.md)
+- [Data model](../specs/260310-seqera-dataset-fs/data-model.md)

docs/reference/cli.md

@@ -1277,11 +1277,10 @@ The `module` command provides a comprehensive system for managing registry-based
 
 (cli-module-run)=
 
-`run [options] [namespace/name] [--<input_name> <input-value>]`
+`run [options] [namespace/name | path] [--<input_name> <input-value>]`
 
-: Execute a module directly from the registry without creating a wrapper workflow.
-: Automatically downloads the module if not already installed. Accepts all standard Nextflow run options.
-: The `module run` command extends the `run` command and accepts all its options, including `-profile`, `-resume`, `-c`, etc. Command-line params (i.e., `--<input_name>`) are inferred from the module's declared inputs.
+: Execute a module directly. Can be a remote module (`namespace/name`) or a local module path (beginning with `./`, `../`, or `/`). Automatically downloads the module if not already installed.
+: Accepts all standard Nextflow run options, including `-profile`, `-resume`, `-c`, etc. Command-line params (i.e., `--<input_name>`) are inferred from the module's declared inputs.
 : The following additional options are available:
 
   `-version`
@@ -1290,15 +1289,20 @@ The `module` command provides a comprehensive system for managing registry-based
 : **Examples:**
 
   ```console
-  # Run module with inputs
-  $ nextflow module run nf-core/fastqc --input 'data/*.fastq.gz'
+  # Run remote module
+  $ nextflow module run nf-core/fastqc \
+      --input 'data/*.fastq.gz'
 
-  # Run specific version with Nextflow options
+  # Run remote module with specific version and run options
   $ nextflow module run nf-core/fastqc \
-      --input 'data/*.fastq.gz' \
       -version 1.0.0 \
-      -profile docker \
+      --input 'data/*.fastq.gz' \
+      -with-conda \
       -resume
+
+  # Run local module
+  $ nextflow module run ./modules/nf-core/fastqc/main.nf \
+      --input 'data/*.fastq.gz'
   ```
 
 (cli-module-search)=

modules/nextflow/src/main/groovy/nextflow/cli/module/CmdModuleRun.groovy

@@ -16,23 +16,22 @@
 
 package nextflow.cli.module
 
+import java.nio.file.Path
+
 import com.beust.jcommander.Parameter
 import com.beust.jcommander.Parameters
 import groovy.transform.CompileStatic
 import io.seqera.npr.client.RegistryClient
+import nextflow.Const
 import nextflow.cli.CmdRun
 import nextflow.config.ConfigBuilder
 import nextflow.config.RegistryConfig
 import nextflow.exception.AbortOperationException
 import nextflow.module.ModuleReference
-import nextflow.module.RegistryClientFactory
 import nextflow.module.ModuleResolver
-
+import nextflow.module.RegistryClientFactory
 import nextflow.util.TestOnly
 
-import java.nio.file.Path
-import java.nio.file.Paths
-
 /**
  * Module run subcommand
  *
@@ -58,31 +57,53 @@ class CmdModuleRun extends CmdRun {
     @Override
     void run() {
         if( !args ) {
-            throw new AbortOperationException("Arguments not provided")
+            throw new AbortOperationException("Module name/path not provided")
+        }
+
+        final moduleFile = isLocalModule(args[0])
+            ? resolveLocalModule(args[0])
+            : resolveRemoteModule(args[0], version)
+
+        if( moduleFile ) {
+            args[0] = moduleFile.toAbsolutePath().toString()
+            super.run()
         }
+    }
+
+    private boolean isLocalModule(String str) {
+        return str.startsWith('/') || str.startsWith('./') || str.startsWith('../')
+    }
+
+    private Path resolveLocalModule(String str) {
+        final module = Path.of(str).toAbsolutePath().normalize()
+        final path = module.isDirectory() ? module.resolve(Const.DEFAULT_MAIN_FILE_NAME) : module
+        if( !path.exists() )
+            throw new AbortOperationException("Invalid module path: ${str}")
+        return path
+    }
 
+    private Path resolveRemoteModule(String name, String version) {
         // Parse and validate module reference
         ModuleReference reference
         try {
-            reference = ModuleReference.parse(args[0])
+            reference = ModuleReference.parse(name)
         } catch( Exception e ) {
-            throw new AbortOperationException("Invalid module reference: ${args[0]}", e)
+            throw new AbortOperationException("Invalid module reference: ${name}", e)
         }
 
         // Get config
-        def baseDir = root ?: Paths.get('.').toAbsolutePath().normalize()
-        def config = new ConfigBuilder()
+        final baseDir = root ?: Path.of('.').toAbsolutePath().normalize()
+        final config = new ConfigBuilder()
             .setOptions(launcher.options)
             .setBaseDir(baseDir)
             .build()
 
-        def registryConfig = config.navigate('registry') as RegistryConfig ?: new RegistryConfig()
-
-        def resolver = new ModuleResolver(baseDir, client ?: RegistryClientFactory.forConfig(registryConfig))
-        Path moduleFile = resolver.installModule(reference, version)
-        if( moduleFile ) {
-            args[0] = moduleFile.toAbsolutePath().toString()
-            super.run()
+        final registryConfig = new RegistryConfig(config.registry as Map ?: Collections.emptyMap())
+        try {
+            final resolver = new ModuleResolver(baseDir, client ?: RegistryClientFactory.forConfig(registryConfig))
+            return resolver.installModule(reference, version)
+        } catch( Exception e ) {
+            throw new AbortOperationException("Unable to install module: ${name}", e)
         }
     }
 }

modules/nextflow/src/test/groovy/nextflow/cli/module/CmdModuleRunTest.groovy

@@ -195,6 +195,73 @@ class CmdModuleRunTest extends Specification {
 
     }
 
+    def 'should run module from local path'() {
+        given:
+        def moduleScript = '''
+            process CREATE_LOCAL_FILE {
+                output:
+                path "local_output.txt"
+
+                script:
+                """
+                echo "Local module executed" > local_output.txt
+                """
+            }
+        '''.stripIndent()
+
+        and:
+        // Create a local module directory with main.nf
+        def moduleDir = tempDir.resolve('local-module')
+        Files.createDirectories(moduleDir)
+        moduleDir.resolve('main.nf').text = moduleScript
+
+        def escapedPath = Pattern.quote(tempDir.toString())
+        def pattern = ~/"${escapedPath}\/.+\/local_output\.txt"/
+
+        and:
+        def cmd = new CmdModuleRun()
+        def opts = new CliOptions()
+        opts.setQuiet(true)
+        cmd.launcher = Mock(Launcher) {
+            getOptions() >> opts
+            getCliString() >> "nextflow module run ${moduleDir}"
+        }
+        cmd.args = [moduleDir.toString()]
+        cmd.root = tempDir
+        cmd.workDir = tempDir.toString()
+        cmd.outputDir = tempDir.resolve('results').toString()
+        cmd.outputFormat = 'json'
+
+        when:
+        cmd.run()
+        def stdout = capture
+            .toString()
+            .readLines()// remove the log part
+            .findResults { line -> !line.contains('DEBUG') ? line : null }
+            .findResults { line -> !line.contains('INFO') ? line : null }.join(" ")
+
+        then:
+        assert (stdout =~ pattern).find()
+    }
+
+    def 'should fail when path and module do not exist'() {
+        given:
+        def nonExistentPath = tempDir.resolve('does-not-exist').toString()
+        def cmd = new CmdModuleRun()
+        cmd.launcher = Mock(Launcher) {
+            getOptions() >> null
+        }
+        cmd.args = [nonExistentPath]
+        cmd.root = tempDir
+
+        when:
+        cmd.run()
+
+        then:
+        def e = thrown(AbortOperationException)
+        e.message.contains('Invalid module path')
+    }
+
     def 'should fail with no arguments'() {
         given:
         def cmd = new CmdModuleRun()

modules/nf-commons/src/main/nextflow/file/FileHelper.groovy

@@ -362,7 +362,7 @@ class FileHelper {
         return asPath(toPathURI(str))
     }
 
-    static final private Map<String,String> PLUGINS_MAP = [s3:'nf-amazon', gs:'nf-google', az:'nf-azure']
+    static final private Map<String,String> PLUGINS_MAP = [s3:'nf-amazon', gs:'nf-google', az:'nf-azure', seqera:'nf-tower']
 
     static final private Map<String,Boolean> SCHEME_CHECKED = new HashMap<>()
 
@@ -373,6 +373,7 @@ class FileHelper {
         // find out the default plugin for the given scheme and try to load it
         final pluginId = PLUGINS_MAP.get(scheme)
         if( pluginId ) try {
+            log.debug "Detected required plugin '$pluginId'"
             if( Plugins.startIfMissing(pluginId) ) {
                 log.debug "Started plugin '$pluginId' required to handle file: $str"
                 // return true to signal a new plugin was loaded

plugins/nf-seqera/src/main/io/seqera/config/ExecutorOpts.groovy

@@ -36,7 +36,7 @@ class ExecutorOpts implements ConfigScope {
     static final Set<String> VALID_AUTO_LABELS = Collections.unmodifiableSet(new LinkedHashSet<>([
         'projectName', 'userName', 'runName', 'sessionId', 'resume',
         'revision', 'commitId', 'repository', 'manifestName',
-        'runtimeVersion', 'workflowId'
+        'runtimeVersion', 'workflowId', 'workspaceId', 'computeEnvId'
     ]))
 
     final RetryOpts retryPolicy
@@ -87,7 +87,7 @@ class ExecutorOpts implements ConfigScope {
             `['runName', 'projectName']` or `'runName,projectName'`
         Valid names: `projectName`, `userName`, `runName`, `sessionId`, `resume`,
         `revision`, `commitId`, `repository`, `manifestName`, `runtimeVersion`,
-        `workflowId`.
+        `workflowId`, `workspaceId`, `computeEnvId`.
     """)
     final Set<String> autoLabels