extend observerv2 with onFileStaged

Signed-off-by: Robrecht Cannoodt <rcannood@gmail.com>

Author: rcannood

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

@@ -77,6 +77,7 @@ import nextflow.trace.TraceObserverV2
 import nextflow.trace.TraceRecord
 import nextflow.trace.WorkflowStatsObserver
 import nextflow.trace.event.FilePublishEvent
+import nextflow.trace.event.FileStagingEvent
 import nextflow.trace.event.TaskEvent
 import nextflow.trace.event.WorkflowOutputEvent
 import nextflow.util.Barrier
@@ -1116,6 +1117,10 @@ class Session implements ISession {
         notifyEvent(observersV2, ob -> ob.onFilePublish(event))
     }
 
+    void notifyFileStaged(FileStagingEvent event) {
+        notifyEvent(observersV2, ob -> ob.onFileStaged(event))
+    }
+
     void notifyFlowComplete() {
         notifyEvent(observersV1, ob -> ob.onFlowComplete())
         notifyEvent(observersV2, ob -> ob.onFlowComplete())

modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy

@@ -39,6 +39,7 @@ import groovy.transform.ToString
 import groovy.util.logging.Slf4j
 import nextflow.Session
 import nextflow.exception.ProcessStageException
+import nextflow.trace.event.FileStagingEvent
 import nextflow.extension.FilesEx
 import nextflow.util.CacheHelper
 import nextflow.util.Duration
@@ -100,6 +101,15 @@ class FilePorter {
         if( batch.size() ) {
             log.trace "Stage foreign files: $batch"
             submitStagingActions(batch.foreignPaths)
+            
+            // Notify observers about file staging completion events
+            for( FileCopy copy : batch.foreignPaths ) {
+                session.notifyFileStaged(new FileStagingEvent(
+                    source: copy.source,
+                    target: copy.target
+                ))
+            }
+            
             log.trace "Stage foreign files completed: $batch"
         }
     }

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

@@ -21,6 +21,7 @@ import groovy.transform.CompileStatic
 import nextflow.Session
 import nextflow.processor.TaskProcessor
 import nextflow.trace.event.FilePublishEvent
+import nextflow.trace.event.FileStagingEvent
 import nextflow.trace.event.TaskEvent
 import nextflow.trace.event.WorkflowOutputEvent
 
@@ -135,4 +136,11 @@ interface TraceObserverV2 {
      */
     default void onFilePublish(FilePublishEvent event) {}
 
+    /**
+     * Invoked when a file staging operation completes (after the file has been copied).
+     *
+     * @param event
+     */
+    default void onFileStaged(FileStagingEvent event) {}
+
 }

modules/nextflow/src/main/groovy/nextflow/trace/event/FileStagingEvent.groovy

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2013-2024, 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.trace.event
+
+import java.nio.file.Path
+
+import groovy.transform.Canonical
+import groovy.transform.CompileStatic
+
+/**
+ * Models a file staging event.
+ *
+ * @author Robrecht Cannoodt <robrecht.cannoodt@gmail.com>
+ */
+@Canonical
+@CompileStatic
+class FileStagingEvent {
+    /**
+     * The original source path (e.g., remote URL or path).
+     */
+    Path source
+    /**
+     * The target staged path in the work directory.
+     */
+    Path target
+}