pool: add firefly onStart marker for MoverProtocol-based transfers

AbstractMoverProtocolTransferService, which handles remote/TPC transfers
(e.g., RemoteHttpTransferService for FTS third-party copies), never
called transferLifeCycle.onStart().  This meant that firefly flow-start
UDP markers were never emitted for TPC movers, even though onEnd markers
were correctly sent via DefaultPostTransferService.

Only NettyTransferService (direct client connections) called onStart(),
so pool-side SciTag flow markers for TPC transfers were incomplete:
ESnet Stardust dashboards received end markers with correct experiment
and activity IDs from transfer tags, but no corresponding start markers.

Add a startTransferLifeCycle() call in MoverTask.run() that derives the
local endpoint from the remote address using NetworkUtils.getLocalAddress
(the same approach used by DefaultPostTransferService.deriveLocalEndpoint
for onEnd) and invokes transferLifeCycle.onStart() before I/O begins.

Motivation: ATLAS AGLT2 pool logs showed 194 TPC onEnd markers with
classifier=transfer-tag and correct activityIds (11, 15, 16) on a
typical day, but zero TPC onStart markers.  Direct-read start markers
(from NettyTransferService) were present but all carried activityId=1
(FQAN fallback) since worker nodes do not send SciTag headers.

Signed-off-by: Shawn McKee <smckee@umich.edu>

Author: ShawnMcKee

modules/dcache/src/main/java/org/dcache/pool/classic/AbstractMoverProtocolTransferService.java

@@ -1,6 +1,6 @@
 /* dCache - http://www.dcache.org/
  *
- * Copyright (C) 2015 - 2025 Deutsches Elektronen-Synchrotron
+ * Copyright (C) 2015 - 2026 Deutsches Elektronen-Synchrotron
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as
@@ -20,6 +20,7 @@
 import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import diskCacheV111.util.CacheException;
+import diskCacheV111.vehicles.IpProtocolInfo;
 import diskCacheV111.vehicles.PoolIoFileMessage;
 import diskCacheV111.vehicles.ProtocolInfo;
 import dmg.cells.nucleus.AbstractCellComponent;
@@ -29,6 +30,9 @@
 import java.io.InterruptedIOException;
 import java.io.SyncFailedException;
 import java.lang.reflect.InvocationTargetException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.CompletionHandler;
 import java.nio.file.StandardOpenOption;
@@ -43,6 +47,7 @@
 import org.dcache.pool.repository.RepositoryChannel;
 import org.dcache.util.CDCExecutorServiceDecorator;
 import org.dcache.util.Exceptions;
+import org.dcache.util.NetworkUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Required;
@@ -143,6 +148,7 @@ public MoverTask(MoverProtocolMover mover,
         public void run() {
             try {
                 setThread();
+                startTransferLifeCycle();
                 RepositoryChannel fileIoChannel = _mover.openChannel();
                 try {
                     if (_mover.getIoMode().contains(StandardOpenOption.WRITE)) {
@@ -183,6 +189,53 @@ public void run() {
             }
         }
 
+        private void startTransferLifeCycle() {
+            if (_transferLifeCycle == null) {
+                SCITAGS_LOGGER.debug(
+                      "scitags lifecycle=start skip reason=no-transfer-lifecycle protocol={} pnfsid={} transferTag={}",
+                      protocolName(),
+                      _mover.getFileAttributes().getPnfsId(),
+                      transferTag());
+                return;
+            }
+
+            if (!(_mover.getProtocolInfo() instanceof IpProtocolInfo ipProtocolInfo)) {
+                SCITAGS_LOGGER.debug(
+                      "scitags lifecycle=start skip reason=non-ip-protocol protocol={} pnfsid={} transferTag={}",
+                      protocolName(),
+                      _mover.getFileAttributes().getPnfsId(),
+                      transferTag());
+                return;
+            }
+
+            InetSocketAddress remoteEndpoint = ipProtocolInfo.getSocketAddress();
+            if (remoteEndpoint == null) {
+                SCITAGS_LOGGER.debug(
+                      "scitags lifecycle=start skip reason=no-remote-endpoint protocol={} pnfsid={} transferTag={}",
+                      protocolName(),
+                      _mover.getFileAttributes().getPnfsId(),
+                      transferTag());
+                return;
+            }
+
+            try {
+                InetAddress localAddress =
+                      NetworkUtils.getLocalAddress(remoteEndpoint.getAddress());
+                InetSocketAddress localEndpoint =
+                      new InetSocketAddress(localAddress, 0);
+                _transferLifeCycle.onStart(remoteEndpoint, localEndpoint,
+                      _mover.getProtocolInfo(), _mover.getSubject());
+            } catch (SocketException e) {
+                SCITAGS_LOGGER.debug(
+                      "scitags lifecycle=start skip reason=local-endpoint-derivation-failed protocol={} pnfsid={} remote={} transferTag={} message={}",
+                      protocolName(),
+                      _mover.getFileAttributes().getPnfsId(),
+                      remoteEndpoint,
+                      transferTag(),
+                      e.getMessage());
+            }
+        }
+
         private void handleChecksumMover() {
             MoverProtocol mover = _mover.getMover();