Add TaskReadinessGate plugin extension point

docs/developer/task-readiness-gate.md

@@ -0,0 +1,91 @@
+(task-readiness-gate)=
+
+# `TaskReadinessGate`
+
+`TaskReadinessGate` is a plugin extension point that defers task submission until an external precondition is met — for example, restoring an S3 object from Glacier before an AWS Batch worker tries to stage it. It works uniformly with every executor that uses Nextflow's `TaskPollingMonitor` and removes the need for plugins to subclass an executor and its task handler purely to override `TaskHandler.isReady()`.
+
+## Interface
+
+A gate implements one method:
+
+```groovy
+package nextflow.processor
+
+import org.pf4j.ExtensionPoint
+
+interface TaskReadinessGate extends ExtensionPoint {
+    void prepare(TaskHandler handler) throws InterruptedException
+}
+```
+
+The plugin registers an implementation via the standard PF4J `@Extension` mechanism, the same way `TraceObserverFactory`, `CacheFactory`, and other extension points are discovered.
+
+## Contract
+
+- **Blocking is allowed.** `prepare` runs on a managed virtual-thread executor inside `TaskPollingMonitor`. Calling `Thread.sleep`, blocking I/O, or long-polling APIs is fine. The scheduler thread is never blocked.
+- **Throwing fails the task.** Any exception marks the task as permanently failed and routes the cause through the task's `errorStrategy` directive. `ProcessException` (and subclasses) propagate identity-preserved. Other throwables are wrapped in a `ProcessException` with the original attached as `cause`, so retry markers like `ProcessRetryableException` reach `TaskProcessor.resumeOrDie` intact.
+- **Retry markers** (`ProcessRetryableException`, `CloudSpotTerminationException`) are recognised by `resumeOrDie` on the *cause* of the thrown exception, not on the exception itself. If you want `errorStrategy 'retry'` to fire for a transient failure, throw the marker as-is (it will be wrapped) — do not pre-wrap it in a `ProcessException`, since the `ProcessException` would propagate identity-preserved and the marker would be lost.
+- **Interrupts must be honored.** Task eviction, workflow abort, and the `executor.gateMaxWait` safety net cancel the in-flight `prepare` by interrupting its thread. Use interruptible primitives (`Thread.sleep`, blocking I/O on NIO channels, `Future.get`).
+- **Multiple gates compose.** When several plugins register gates, all must complete successfully before the task is admitted. Gates run in parallel; order is unspecified.
+
+## Configuration
+
+`executor.gateMaxWait` bounds the time a gate may spend preparing a single task (default `24h`). Tasks whose gates exceed this limit are cancelled and fail with a `ProcessException` that the task's `errorStrategy` can handle (including `retry`).
+
+## Per-process opt-out
+
+Use the existing `hints` directive — no core change required:
+
+```nextflow
+process FAST_PATH {
+    hints 'glacier/skip': true
+    // ...
+}
+```
+
+Inside the gate, inspect `handler.task.config.hints` and short-circuit:
+
+```groovy
+@Override
+void prepare(TaskHandler handler) throws InterruptedException {
+    if( handler.task.config.hints['glacier/skip'] == true ) return
+    // ... real work
+}
+```
+
+The `hints` directive uses dot-separated and prefix-separated keys; plugins should namespace their hint keys (e.g. `glacier/skip`, `mycorp.cold-storage/skip`) to avoid collisions.
+
+## Example
+
+A minimal gate that issues an S3 Glacier restore request and waits for completion:
+
+```groovy
+@CompileStatic
+class GlacierReadinessGate implements TaskReadinessGate {
+
+    private final GlacierRestoreManager manager
+
+    GlacierReadinessGate() {
+        this.manager = new GlacierRestoreManager(/* config from session */)
+    }
+
+    @Override
+    void prepare(TaskHandler handler) throws InterruptedException {
+        if( handler.task.config.hints['glacier/skip'] == true ) return
+        for( S3Path path : extractS3Inputs(handler.task) ) {
+            manager.issueRestoreIfNeeded(path)        // idempotent
+            while( !manager.isRestored(path) ) {
+                if( manager.isPermanentlyFailed(path) )
+                    throw new ProcessException("Glacier restore failed for ${path}")
+                Thread.sleep(60_000)                  // virtual thread parks
+            }
+        }
+    }
+}
+```
+
+## When `prepare` runs
+
+`TaskPollingMonitor.schedule()` submits `prepare` to the managed executor the moment the task is enqueued — before any executor slot has freed up, before `canForkProcess()` is consulted. This means restore work for tasks queued behind a full executor begins immediately, not when slots free up.
+
+`canSubmit()` then polls the resulting `Future` on every monitor tick. The task is admitted as soon as every gate's future has completed successfully and the standard `canForkProcess` / `isReady` / capacity checks pass.

docs/reference/config.md

@@ -751,6 +751,9 @@ The following settings are available:
 : *Used only by grid executors.*
 : Determines how long to wait for the `.exitcode` file to be created after the task has completed, before returning an error status (default: `270 sec`).
 
+`executor.gateMaxWait`
+: Maximum time a `TaskReadinessGate` plugin may take to prepare a task before the task is failed (default: `24h`). See the [TaskReadinessGate developer page](../developer/task-readiness-gate.md) for details.
+
 `executor.jobName`
 : *Used only by grid executors and Google Batch.*
 : Determines the name of jobs submitted to the underlying cluster executor:

modules/nextflow/src/main/groovy/nextflow/executor/ExecutorConfig.groovy

@@ -61,6 +61,12 @@ class ExecutorConfig implements ConfigScope {
     """)
     final Duration exitReadTimeout
 
+    @ConfigOption
+    @Description("""
+        Maximum time a `TaskReadinessGate` plugin may take to prepare a task before the task is failed (default: `24h`).
+    """)
+    final Duration gateMaxWait
+
     @ConfigOption
     @Description('''
         *Used only by grid executors and Google Batch.*
@@ -173,6 +179,7 @@ class ExecutorConfig implements ConfigScope {
         cpus = opts.cpus as Integer
         dumpInterval = opts.dumpInterval as Duration ?: Duration.of('5min')
         exitReadTimeout = opts.exitReadTimeout as Duration ?: Duration.of('270sec')
+        gateMaxWait = opts.gateMaxWait as Duration ?: Duration.of('24h')
         jobName = opts.jobName as Closure
         killBatchSize = opts.killBatchSize != null ? opts.killBatchSize as int : 100
         memory = opts.memory as MemoryUnit

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -18,7 +18,13 @@ package nextflow.processor
 
 import static nextflow.processor.TaskProcessor.*
 
+import java.util.concurrent.Callable
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentMap
+import java.util.concurrent.ExecutionException
 import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.Future
 import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.locks.Condition
@@ -27,6 +33,7 @@ import java.util.concurrent.locks.ReentrantLock
 
 import com.google.common.util.concurrent.RateLimiter
 import groovy.transform.CompileStatic
+import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import nextflow.Session
 import nextflow.SysEnv
@@ -39,6 +46,8 @@ import nextflow.exception.ProcessSubmitTimeoutException
 import nextflow.executor.BatchCleanup
 import nextflow.executor.ExecutorConfig
 import nextflow.executor.GridTaskHandler
+import nextflow.plugin.Plugins
+import nextflow.util.CustomThreadFactory
 import nextflow.util.Duration
 import nextflow.util.SysHelper
 import nextflow.util.Threads
@@ -132,6 +141,36 @@ class TaskPollingMonitor implements TaskMonitor {
 
     private boolean enableAsyncFinalizer = SysEnv.getBool('NXF_ENABLE_ASYNC_FINALIZER',true)
 
+    @PackageScope
+    List<TaskReadinessGate> readinessGates = Collections.emptyList()
+
+    /**
+     * Background executor on which {@link TaskReadinessGate#prepare} runs. Must be a
+     * non-blocking-submission executor (virtual-thread or unbounded cached pool) so that
+     * {@link #schedule} never blocks the caller while holding {@code pendingLock}.
+     */
+    @PackageScope
+    ExecutorService gateExecutor
+
+    @PackageScope
+    final ConcurrentMap<TaskHandler, GateState> gateStates = new ConcurrentHashMap<>()
+
+    @PackageScope
+    Duration gateMaxWait
+
+    /**
+     * Tracks in-flight {@link TaskReadinessGate#prepare} futures for a single handler,
+     * together with the wall-clock timestamp at which the task was scheduled. The
+     * timestamp is used by the {@code executor.gateMaxWait} safety net to decide when
+     * a stuck gate should be cancelled and the task failed.
+     */
+    @PackageScope
+    static class GateState {
+        final long scheduledAt = System.currentTimeMillis()
+        final List<Future<?>> futures
+        GateState(List<Future<?>> futures) { this.futures = futures }
+    }
+
     /**
      * Create the task polling monitor with the provided named parameters object.
      * <p>
@@ -154,12 +193,20 @@ class TaskPollingMonitor implements TaskMonitor {
         this.name = params.name
         this.session = params.session as Session
         this.config = params.config as ExecutorConfig
+        this.gateMaxWait = config?.gateMaxWait
         this.pollIntervalMillis = ( params.pollInterval as Duration ).toMillis()
         this.dumpInterval = params.dumpInterval as Duration
         this.capacity = (params.capacity ?: 0) as int
 
         this.pendingQueue = new LinkedBlockingQueue<TaskHandler>()
         this.runningQueue = new LinkedBlockingQueue<TaskHandler>()
+
+        this.taskCompleteLock = new ReentrantLock()
+        this.taskComplete = taskCompleteLock.newCondition()
+
+        this.pendingLock = new ReentrantLock()
+        this.taskAvail = pendingLock.newCondition()
+        this.slotAvail = pendingLock.newCondition()
     }
 
     static TaskPollingMonitor create( Session session, ExecutorConfig config, String name, int defQueueSize, Duration defPollInterval ) {
@@ -248,7 +295,51 @@ class TaskPollingMonitor implements TaskMonitor {
      *      by the polling monitor
      */
     protected boolean canSubmit(TaskHandler handler) {
-        (capacity > 0 ? checkQueueCapacity(handler) : true) && handler.canForkProcess() && handler.isReady()
+        allGatesReady(handler) \
+            && handler.canForkProcess() \
+            && handler.isReady() \
+            && (capacity > 0 ? checkQueueCapacity(handler) : true)
+    }
+
+    private boolean allGatesReady(TaskHandler handler) {
+        final state = gateStates.get(handler)
+        if( !state ) return true
+
+        if( gateMaxWait && System.currentTimeMillis() - state.scheduledAt > gateMaxWait.toMillis() ) {
+            state.futures*.cancel(true)
+            gateStates.remove(handler)
+            throw new ProcessException("Task readiness gate timed out after ${gateMaxWait} for task '${handler.task.name}'")
+        }
+
+        for( f in state.futures ) {
+            if( !f.isDone() ) return false
+            if( f.isCancelled() ) {
+                state.futures*.cancel(true)
+                gateStates.remove(handler)
+                throw new ProcessException("Task readiness gate was cancelled for task '${handler.task.name}'")
+            }
+            try { f.get() }
+            catch( ExecutionException e ) {
+                // cancel peer gates so their work doesn't outlive the failing task
+                state.futures*.cancel(true)
+                gateStates.remove(handler)
+                // Rethrow ProcessException (and subclasses) identity-preserved so the
+                // original message and type reach resumeOrDie. Wrap anything else in a
+                // ProcessException with the original attached as cause — this preserves
+                // ProcessRetryableException routing in resumeOrDie, which inspects
+                // `error.cause`, not `error` itself.
+                final cause = e.cause
+                if( cause instanceof ProcessException )
+                    throw (ProcessException) cause
+                throw new ProcessException("Task readiness gate failed for task '${handler.task.name}'", cause ?: e)
+            }
+            catch( InterruptedException e ) {
+                Thread.currentThread().interrupt()
+                return false
+            }
+        }
+        gateStates.remove(handler)
+        return true
     }
 
     /**
@@ -305,6 +396,14 @@ class TaskPollingMonitor implements TaskMonitor {
         pendingLock.lock()
         try{
             pendingQueue << handler
+            if( readinessGates ) {
+                final futures = new ArrayList<Future<?>>(readinessGates.size())
+                for( TaskReadinessGate g : readinessGates ) {
+                    final gate = g   // capture in a fresh local for the async closure
+                    futures << gateExecutor.submit({ gate.prepare(handler) } as Callable)
+                }
+                gateStates.put(handler, new GateState(futures))
+            }
             taskAvail.signal()  // signal that a new task is available for execution
             notifyTaskPending(handler)
             log.trace "Scheduled task > $handler"
@@ -329,6 +428,8 @@ class TaskPollingMonitor implements TaskMonitor {
             return false
         }
 
+        gateStates.remove(handler)?.futures*.cancel(true)
+
         if( remove(handler) ) {
             pendingLock.lock()
             try {
@@ -351,16 +452,18 @@ class TaskPollingMonitor implements TaskMonitor {
      */
     @Override
     TaskMonitor start() {
+        readinessGates = Plugins.getExtensions(TaskReadinessGate)
+        if( readinessGates ) {
+            gateExecutor = Threads.useVirtual()
+                ? Executors.newVirtualThreadPerTaskExecutor()
+                : Executors.newCachedThreadPool(new CustomThreadFactory('TaskReadinessGate'))
+            session.onShutdown { gateExecutor.shutdownNow() }
+            log.debug "Registered ${readinessGates.size()} task readiness gate(s): ${readinessGates*.class*.simpleName}"
+        }
+
         log.debug ">>> barrier register (monitor: ${this.name})"
         session.barrier.register(this)
 
-        this.taskCompleteLock = new ReentrantLock()
-        this.taskComplete = taskCompleteLock.newCondition()
-
-        this.pendingLock = new ReentrantLock()
-        this.taskAvail = pendingLock.newCondition()
-        this.slotAvail = pendingLock.newCondition()
-
         //
         this.submitRateLimit = createSubmitRateLimit()
 

modules/nextflow/src/main/groovy/nextflow/processor/TaskReadinessGate.groovy

@@ -0,0 +1,49 @@
+/*
+ * 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.processor
+
+import groovy.transform.CompileStatic
+import org.pf4j.ExtensionPoint
+
+/**
+ * Plugin extension point that defers task submission until an external precondition is met.
+ *
+ * <p>Implementations are invoked once per task by the scheduler, on a managed background
+ * thread (virtual thread when available). The task is admitted for submission when this
+ * method returns; throw to mark the task as permanently failed and route the cause through
+ * the task's {@code errorStrategy}.
+ *
+ * <p>Implementations may block freely — {@code Thread.sleep}, network calls, long polling.
+ * The scheduler thread is never blocked by this call.
+ *
+ * <p>Implementations <b>must</b> honor {@code Thread.interrupt()} so that task eviction,
+ * workflow abort, and the {@code executor.gateMaxWait} backstop can unblock {@code prepare}
+ * promptly. Use interruptible primitives ({@code Thread.sleep}, blocking I/O on NIO
+ * channels, {@code Future.get}) and propagate {@code InterruptedException}.
+ *
+ * <p>When multiple gates are registered, a task is admitted only when every gate's
+ * {@code prepare} method has returned successfully. Evaluation order across gates is
+ * unspecified; all gates start in parallel on the managed executor.
+ *
+ * <p>Per-process opt-out is available via the {@code hints} directive — gates that wish
+ * to support it should check a namespaced hint key (e.g. {@code 'glacier/skip': true}) and
+ * return immediately when set. No core mechanism is required.
+ */
+@CompileStatic
+interface TaskReadinessGate extends ExtensionPoint {
+    void prepare(TaskHandler handler) throws InterruptedException
+}

modules/nextflow/src/test/groovy/nextflow/executor/ExecutorConfigTest.groovy

@@ -116,6 +116,20 @@ class ExecutorConfigTest extends Specification {
         config.getExecConfigProp( 'hazelcast', 'jobName', 'alpha', [NXF_EXECUTOR_JOBNAME:'hola']) == 'hola'
     }
 
+    def 'should default gateMaxWait to 24h'() {
+        when:
+        def config = new ExecutorConfig([:])
+        then:
+        config.gateMaxWait == Duration.of('24h')
+    }
+
+    def 'should override gateMaxWait from opts'() {
+        when:
+        def config = new ExecutorConfig(gateMaxWait: '48h')
+        then:
+        config.gateMaxWait == Duration.of('48h')
+    }
+
     def 'test onlyJobState property'() {
 
         when:

modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy

@@ -126,6 +126,8 @@ class ParallelPollingMonitorTest extends Specification {
         and:
         def processor = Mock(TaskProcessor)
         def arrayHandler = Mock(TaskHandler) {
+            canForkProcess() >> true
+            isReady() >> true
             getTask() >> Mock(TaskArrayRun) {
                 getName() >> 'oversized_array'
                 getArraySize() >> 10