Let a plugin supply the task hasher and the cache resolution strategy

Two extension points in nextflow.processor, both with defaults that preserve
today's behaviour exactly:

- TaskHasherFactory creates the TaskHasher for a task, or abstains. With no
  extension registered, or all abstaining, TaskProcessor does `new
  TaskHasher(task)` as before. Core's TaskHasher is untouched.

- TaskCacheStrategy decides how a task is resolved against the cache, through
  the TaskResolver primitives the processor implements. DefaultTaskCacheStrategy
  is master's checkCachedOrLaunchTask loop moved verbatim -- the tries fold, the
  lock, mkdirs, submitTask -- so a run with no extension takes the same path it
  takes today. checkCachedOrLaunchTask becomes the dispatch.

Supporting changes:

- CacheFactory may resolve its own work directory and write it back to the
  session, so Session.init creates the cache before reading workDir. Everything
  below -- the work dir creation, the observers, the WorkflowMetadata snapshot --
  must see the effective value, or workflow.workDir names a directory no task
  uses. A throw after that point now closes the cache rather than leaking it.

- CacheStore.updateEntry, a default method, so a store composed of several
  members can route an update back to the member that served the read.

- CacheDB.dispatchWrite, an internal refactor so a failed async cache write is
  logged with its cause instead of being swallowed.

- FileHelper.getTaskHashFromPath tolerates an optional `-N` suffix on the work
  directory leaf. nf-lineage resolves every task input to its producing task by
  parsing that path (LinObserver.manageFileInParam -> getSourceReference); a
  cache that lays the attempts of one task out side by side keeps the two-level
  <2hex>/<30hex> hierarchy and appends the attempt to the leaf. Without this the
  parser returns null for those inputs and lineage records them as bare
  DataPaths, losing the edge to the producing task and paying a content checksum
  in its place. The parser is already lenient by design -- a non-2-char bucket
  and an unparseable hash both return null -- so this is one more accepted leaf
  shape, and hex never contains a dash, so it stays unambiguous.

Assisted-by: Claude Opus 5 (1M context)
Signed-off-by: jorgee <jorge.ejarque@seqera.io>

Author: jorgee

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -371,6 +371,14 @@ class Session implements ISession {
 
     CacheDB getCache() { cache }
 
+    /**
+     * Open this run's cache. A separate method so a test can substitute a factory that writes back
+     * into the session the way a real one may — see {@link CacheFactory#newInstance}.
+     */
+    protected CacheDB createCache() {
+        return CacheFactory.create(uniqueId, runName).open()
+    }
+
     /**
      * Creates a new session using the configuration properties provided
      *
@@ -461,6 +469,30 @@ class Session implements ISession {
      */
     Session init( ScriptFile scriptFile, List<String> args=null, Map<String,?> cliParams=null, Map<String,?> configParams=null ) {
 
+        // -- create the cache FIRST, and do not read `workDir` above this line: a factory may
+        //    resolve a work dir of its own and write it back here (see CacheFactory.newInstance),
+        //    and everything below -- the work dir creation, the observers, the WorkflowMetadata
+        //    snapshot -- must see the effective value, otherwise `workflow.workDir` reports a
+        //    directory the tasks never use. SessionTest locks this ordering.
+        cache = createCache()
+        try {
+            return init0(scriptFile, args, cliParams, configParams)
+        }
+        catch( Throwable t ) {
+            // Everything below createCache() can throw -- the work dir may be unwritable, an observer
+            // factory may abort -- and ScriptRunner calls init() OUTSIDE the try that later closes the
+            // session, so an abort here would leave the cache open. For the default cache that means a
+            // LevelDB index already truncated by open() and never closed; for a plugin cache it means
+            // whatever that cache holds. Close it and let the original failure propagate.
+            try { cache?.close() }
+            catch( Exception e ) { log.debug "Unable to close the cache after a failed session init -- ${e.message}" }
+            cache = null
+            throw t
+        }
+    }
+
+    private Session init0( ScriptFile scriptFile, List<String> args, Map<String,?> cliParams, Map<String,?> configParams ) {
+
         if(!workDir.mkdirs())
             throw new AbortOperationException("Cannot create work-dir '${FilesEx.toUriString(workDir)}' -- Make sure you have write permissions or specify a different directory by using the `-w` command line option")
         log.debug "Work-dir: ${workDir.toUriString()} [${FileHelper.getPathFsType(workDir)}]"
@@ -492,8 +524,6 @@ class Session implements ISession {
         binding.setParams( (Map)config.params )
         binding.setArgs( new ScriptRunner.ArgsList(args) )
 
-        cache = CacheFactory.create(uniqueId,runName).open()
-
         return this
     }
 

modules/nextflow/src/main/groovy/nextflow/cache/CacheDB.groovy

@@ -16,7 +16,6 @@
 
 package nextflow.cache
 
-
 import com.google.common.hash.HashCode
 import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
@@ -85,10 +84,45 @@ class CacheDB implements Closeable {
         final record = (List)KryoHelper.deserialize(payload)
         TraceRecord trace = TraceRecord.deserialize( (byte[])record[0] )
         TaskContext ctx = record[1]!=null && processor!=null ? TaskContext.deserialize(processor, (byte[])record[1]) : null
-
         return new TaskEntry(trace,ctx)
     }
 
+    /**
+     * Dispatch an asynchronous cache write on the {@link #writer} agent, logging any failure
+     * instead of letting the agent swallow it. A dropped entry / index write
+     * would otherwise leave a successfully executed task non-resumable with no trace: the run still
+     * succeeds, but the task re-executes on the next resume. Surfacing the error makes that visible
+     * (and is the natural hook for a future retry at the store level).
+     *
+     * @param what   short description of the write for the log message (typically the task hash)
+     * @param action the store mutation to run on the writer thread
+     */
+    protected void dispatchWrite(String what, Closure action) {
+        writer.send {
+            try {
+                action.call()
+            }
+            catch( Throwable e ) {
+                log.warn("Unable to persist cache record for ${what} -- the task will re-execute on the next resume", e)
+            }
+        }
+    }
+
+    /**
+     * Bump the reference count of an entry that already exists, which also refreshes the stored
+     * object's last-modified stamp.
+     *
+     * <b>The count is advisory for a shared store.</b> This is a read-modify-write with no
+     * compare-and-set behind {@link CacheStore}, so two runs resuming the same task can read the same
+     * value and write the same increment, losing one. That is harmless today only because neither
+     * consumer of the count is reachable for a shared cache: {@link #removeTaskEntry} — the only
+     * decrement, and the only path that can delete on reaching zero — is called from
+     * {@code Session.cleanup}, which returns before opening the cache for a non-{@code file:} work
+     * dir, and from {@code CmdClean}, which a shared cache's {@code CacheDB} may refuse by
+     * overriding it; and a cross-run cache ages entries by the object's last-modified time, not by
+     * the count. Eviction tooling that reads the count instead would need a genuinely atomic update
+     * here.
+     */
     void incTaskEntry( HashCode hash ) {
         final payload = store.getEntry(hash)
         if( !payload ) {
@@ -99,11 +133,15 @@ class CacheDB implements Closeable {
         final record = (List)KryoHelper.deserialize(payload)
         // third record contains the reference count for this record
         record[2] = ((Integer)record[2]) +1
-        // save it again
-        store.putEntry(hash, KryoHelper.serialize(record))
+        // save it again -- an update of the record just read, not a new entry (see updateEntry)
+        store.updateEntry(hash, KryoHelper.serialize(record))
 
     }
 
+    /**
+     * Decrement the reference count, deleting the entry when it reaches zero. Callers must not invoke
+     * this on a shared cache — see the advisory-count note on {@link #incTaskEntry}.
+     */
     boolean removeTaskEntry( HashCode hash ) {
         final payload = store.getEntry(hash)
         if( !payload ) {
@@ -114,9 +152,9 @@ class CacheDB implements Closeable {
         final record = (List)KryoHelper.deserialize(payload)
         // third record contains the reference count for this record
         def count = record[2] = ((Integer)record[2]) -1
-        // save or delete
+        // save or delete -- as in incTaskEntry, saving is an update of the record just read
         if( count > 0 ) {
-            store.putEntry(hash, KryoHelper.serialize(record))
+            store.updateEntry(hash, KryoHelper.serialize(record))
             return false
         }
         else {
@@ -153,18 +191,18 @@ class CacheDB implements Closeable {
     }
 
     void putTaskAsync( TaskHandler handler, TraceRecord trace ) {
-        writer.send { writeTaskEntry0(handler, trace) }
+        dispatchWrite("task entry ${handler.task.hash}") { writeTaskEntry0(handler, trace) }
     }
 
     void cacheTaskAsync( TaskHandler handler ) {
-        writer.send {
+        dispatchWrite("cached task ${handler.task.hash}") {
             writeTaskIndex0(handler,true)
             incTaskEntry(handler.task.hash)
         }
     }
 
     void putIndexAsync(TaskHandler handler ) {
-        writer.send { writeTaskIndex0(handler) }
+        dispatchWrite("task index ${handler.task.hash}") { writeTaskIndex0(handler) }
     }
 
     @PackageScope

modules/nextflow/src/main/groovy/nextflow/cache/CacheFactory.groovy

@@ -32,6 +32,20 @@ import org.pf4j.ExtensionPoint
 @CompileStatic
 abstract class CacheFactory implements ExtensionPoint {
 
+    /**
+     * Build the cache instance.
+     *
+     * <p>A factory MAY write back into the current session while resolving its cache — a
+     * content-addressable cache whose work directory <b>is</b> the cache assigns
+     * {@code session.workDir}, and turns {@code resumeMode} on because its hits are keyed by task
+     * hash rather than by session id. This is why {@code Session.init} creates the cache before it
+     * reads {@code workDir}: everything downstream of that point (the work-dir creation, the
+     * observers, the {@code WorkflowMetadata} snapshot) must see the effective value, or
+     * {@code workflow.workDir} reports a directory the tasks never use.
+     *
+     * <p>{@code SessionTest} locks that ordering, so a reorder fails a test rather than silently
+     * producing a run whose reported work dir is not the one in use.
+     */
     protected abstract CacheDB newInstance(UUID uniqueId, String runName, Path home=null)
 
     static CacheDB create(UUID uniqueId, String runName, Path home=null) {

modules/nextflow/src/main/groovy/nextflow/cache/CacheStore.groovy

@@ -41,6 +41,18 @@ interface CacheStore {
     void putEntry(HashCode key, byte[] value)
     void deleteEntry(HashCode key)
 
+    /**
+     * Update an entry that already exists in this store, i.e. the read-modify-write of the
+     * reference count / last-used stamp in {@link CacheDB#incTaskEntry}, whose value is a record
+     * just read back through {@link #getEntry}.
+     *
+     * This is deliberately distinct from {@link #putEntry}, which stores a <b>new</b> entry: a
+     * composite store has to send an update to the member that served the read — and may have to
+     * drop it when that member is read-only — while a new entry must always go to the writable
+     * one. Defaults to {@link #putEntry}, which is the correct behaviour for a single store.
+     */
+    default void updateEntry(HashCode key, byte[] value) { putEntry(key, value) }
+
     void writeIndex(HashCode key, boolean cached)
     Iterator<Index> iterateIndex()
     void deleteIndex()

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

@@ -0,0 +1,105 @@
+/*
+ * 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 java.nio.file.Path
+
+import com.google.common.hash.HashCode
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import nextflow.Session
+import nextflow.file.FileHelper
+import nextflow.util.HashBuilder
+import nextflow.util.LockManager
+
+/**
+ * The default {@link TaskCacheStrategy}: the per-run, local resolution loop as it has always been in
+ * {@code TaskProcessor.checkCachedOrLaunchTask}, relocated here unchanged.
+ *
+ * <p>The task hash is folded with the attempt number ({@code tries}) into the per-attempt hash that
+ * names the work directory and keys the cache entry. For each attempt the entry is looked up and, when
+ * resuming, the task is resumed from it; an attempt whose work directory already exists -- a previous
+ * failure, or an identical task instance of this run -- bumps to the next one; the first free work
+ * directory is created under an in-process lock and the task is launched there.
+ *
+ * <p>This strategy is not a plugin extension: it is the fallback {@link TaskProcessor} uses when no
+ * registered {@link TaskCacheStrategy} applies to the session, so {@link #isEnabled} is always true.
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+@Slf4j
+@CompileStatic
+class DefaultTaskCacheStrategy implements TaskCacheStrategy {
+
+    private static LockManager lockManager = new LockManager()
+
+    @Override
+    boolean isEnabled(Session session) { return true }
+
+    @Override
+    void resolve(TaskRun task, HashCode hash, boolean shouldTryCache, TaskResolver resolver) {
+
+        int tries = task.failCount +1
+        while( true ) {
+            hash = HashBuilder.defaultHasher().putBytes(hash.asBytes()).putInt(tries).hash()
+
+            Path resumeDir = null
+            boolean exists = false
+            try {
+                final entry = resolver.entry(hash)
+                resumeDir = entry ? FileHelper.asPath(entry.trace.getWorkDir()) : null
+                if( resumeDir )
+                    exists = resumeDir.exists()
+
+                log.trace "[${task.lazyName()}] Cacheable folder=${resumeDir?.toUriString()} -- exists=$exists; try=$tries; shouldTryCache=$shouldTryCache; entry=$entry"
+                final cached = shouldTryCache && exists && entry.trace.isCompleted() && resolver.resume(task, hash, resumeDir, entry)
+                if( cached )
+                    break
+            }
+            catch (Throwable t) {
+                log.warn1("[${task.lazyName()}] Unable to resume cached task -- See log file for details", causedBy: t)
+            }
+
+            if( exists ) {
+                tries++
+                continue
+            }
+
+            final lock = lockManager.acquire(hash)
+            final workDir = resolver.workDirFor(hash)
+            try {
+                if( resumeDir != workDir )
+                    exists = workDir.exists()
+                if( exists ) {
+                    tries++
+                    continue
+                }
+                else if( !workDir.mkdirs() )
+                    throw new IOException("Unable to create directory=$workDir -- check file system permissions")
+            }
+            finally {
+                lock.release()
+            }
+
+            // submit task for execution
+            resolver.launch( task, hash, workDir )
+            break
+        }
+
+    }
+
+}

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

@@ -0,0 +1,62 @@
+/*
+ * 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 com.google.common.hash.HashCode
+import nextflow.Session
+import org.pf4j.ExtensionPoint
+
+/**
+ * Plugin extension point deciding how a task is resolved against the cache: resumed from a previous
+ * execution, or given a work directory and launched.
+ *
+ * <p>{@link TaskProcessor} resolves the registered strategies once, in {@link nextflow.plugin.Priority}
+ * order, and uses the first one that {@link #isEnabled applies} to the session; with none registered,
+ * or none applying, the {@link DefaultTaskCacheStrategy} is used, so a run without such a plugin
+ * resolves its tasks exactly as before.
+ *
+ * <p>A strategy is expected to abstain (return {@code false} from {@link #isEnabled}) whenever its cache
+ * is not the one in use for the session: the plugin registry is process-wide, while the choice of
+ * strategy belongs to the run.
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+interface TaskCacheStrategy extends ExtensionPoint {
+
+    /**
+     * Whether this strategy applies to the run; the highest-priority applicable one wins, else the
+     * default.
+     *
+     * @param session The session of the run.
+     * @return {@code true} to take over the resolution of every task of the run.
+     */
+    boolean isEnabled(Session session)
+
+    /**
+     * Resolve {@code task}, whose hash is {@code hash}: end by calling exactly one of
+     * {@link TaskResolver#resume resolver.resume} (returning {@code true}) or
+     * {@link TaskResolver#launch resolver.launch}.
+     *
+     * @param task The task to resolve.
+     * @param hash The task hash, as computed by its {@link TaskHasher}.
+     * @param tryCache Whether a cached execution may be resumed; {@code false} on a retry, and when
+     *      the run is not resuming.
+     * @param resolver The processor's primitives to resume or launch the task with.
+     */
+    void resolve(TaskRun task, HashCode hash, boolean tryCache, TaskResolver resolver)
+
+}