Fix ResourcesAggregator deadlock with virtual thread executor (#6840)

Use a dedicated fixed thread pool in computeSummaryMap() instead of
the shared session executor to prevent starvation deadlock when
virtual thread carrier threads are saturated.

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: pditommaso

modules/nextflow/src/main/groovy/nextflow/trace/ReportObserver.groovy

@@ -113,7 +113,7 @@ class ReportObserver implements TraceObserverV2 {
     @Override
     void onFlowCreate(Session session) {
         this.session = session
-        this.aggregator = new ResourcesAggregator(session)
+        this.aggregator = new ResourcesAggregator()
         // check if the process exists
         if( Files.exists(reportFile) && !overwrite )
             throw new AbortOperationException("Report file already exists: ${reportFile.toUriString()} -- enable the 'report.overwrite' option in your config file to overwrite existing files")

modules/nextflow/src/main/groovy/nextflow/trace/ResourcesAggregator.groovy

@@ -1,17 +1,16 @@
 package nextflow.trace
 
 import java.util.concurrent.Callable
-import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
 import java.util.concurrent.Future
 
 import groovy.json.JsonOutput
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
-import nextflow.Session
 
 /**
  * Collect and aggregate execution metrics used by execution report
- * 
+ *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
 @Slf4j
@@ -23,15 +22,6 @@ class ResourcesAggregator {
      */
     final private Map<String,ReportSummary> summaries = new LinkedHashMap<>()
 
-    /**
-     * Executor service used to compute report summary stats
-     */
-    private ExecutorService executor
-
-    ResourcesAggregator(Session session) {
-        this.executor = session.getExecService()
-    }
-
     /**
      * Aggregates task record for each process in order to render the
      * final execution stats
@@ -71,16 +61,24 @@ class ResourcesAggregator {
             result.put(process, new HashMap(10))
         }
 
-        // submit the parallel execution
-        final allResults = executor.invokeAll(tasks)
-
-        // compose the final result
-        for( Future<List> future : allResults ) {
-            final triple = future.get()
-            final name = triple[0]      // the process name
-            final series = triple[1]    // the series name eg. `cpu`, `time`, etc
-            final summary = triple[2]   // the computed summary
-            result.get(name).put(series, summary)
+        // use a dedicated thread pool to avoid starvation deadlock
+        // with the shared virtual thread executor (see #6833)
+        final executor = Executors.newFixedThreadPool(Runtime.runtime.availableProcessors())
+        try {
+            // submit the parallel execution
+            final allResults = executor.invokeAll(tasks)
+
+            // compose the final result
+            for( Future<List> future : allResults ) {
+                final triple = future.get()
+                final name = triple[0]      // the process name
+                final series = triple[1]    // the series name eg. `cpu`, `time`, etc
+                final summary = triple[2]   // the computed summary
+                result.get(name).put(series, summary)
+            }
+        }
+        finally {
+            executor.shutdown()
         }
 
         return result

modules/nextflow/src/test/groovy/nextflow/trace/ResourcesAggregatorTest.groovy

@@ -2,10 +2,7 @@ package nextflow.trace
 
 import spock.lang.Specification
 
-import java.util.concurrent.Executors
-
 import groovy.json.JsonSlurper
-import nextflow.Session
 
 /**
  *
@@ -21,12 +18,7 @@ class ResourcesAggregatorTest extends Specification {
 
     def 'should render summary json' () {
         given:
-        def executor = Executors.newCachedThreadPool()
-        def session = Mock(Session) {
-            getExecService() >> executor
-        }
-
-        def observer = new ResourcesAggregator(session)
+        def observer = new ResourcesAggregator()
         observer.aggregate(r1)
         observer.aggregate(r2)
         observer.aggregate(r3)
@@ -119,21 +111,13 @@ class ResourcesAggregatorTest extends Specification {
         result[1].timeUsage.q2 == 102.50
         result[1].timeUsage.q3 == 108.75
         result[1].timeUsage.mean == 102.50
-
-        cleanup:
-        observer?.executor?.shutdown()
     }
 
 
     def 'should compute summary list' () {
 
         given:
-        def executor = Executors.newCachedThreadPool()
-        def session = Mock(Session) {
-            getExecService() >> executor
-        }
-
-        def observer = new ResourcesAggregator(session)
+        def observer = new ResourcesAggregator()
         observer.aggregate(r1)
         observer.aggregate(r2)
         observer.aggregate(r3)
@@ -156,20 +140,12 @@ class ResourcesAggregatorTest extends Specification {
         result[1].mem."max" == 22000.0
         result[1].time.min == 18000
         result[1].time.max == 23000
-
-        cleanup:
-        observer?.executor?.shutdown()
     }
 
 
     def 'should maintain insertion order' () {
         given:
-        def executor = Executors.newCachedThreadPool()
-        def session = Mock(Session) {
-            getExecService() >> executor
-        }
-
-        def observer = new ResourcesAggregator(session)
+        def observer = new ResourcesAggregator()
         observer.aggregate(new TraceRecord([process: 'gamma', name: 'gamma-1']))
         observer.aggregate(new TraceRecord([process: 'delta', name: 'delta-1']))
         observer.aggregate(new TraceRecord([process: 'delta', name: 'delta-2']))

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

@@ -271,7 +271,7 @@ class TowerClient implements TraceObserverV2 {
         log.debug "Creating Seqera Platform observer -- endpoint=$endpoint; requestInterval=$requestInterval; aliveInterval=$aliveInterval; maxRetries=$maxRetries; backOffBase=$backOffBase; backOffDelay=$backOffDelay"
 
         this.session = session
-        this.aggregator = new ResourcesAggregator(session)
+        this.aggregator = new ResourcesAggregator()
         this.runName = session.getRunName()
         this.runId = session.getUniqueId()
         this.httpClient = newHttpClient()