Make transport-window uploads acknowledgement-safe

src/main/java/com/gpuflight/agent/AcknowledgedWindowCleaner.java

@@ -0,0 +1,147 @@
+package com.gpuflight.agent;
+
+import com.gpuflight.agent.config.JsonSettings;
+import com.gpuflight.agent.model.WindowMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Deletes a transport payload only after its backend ACK while retaining the
+ * immutable metadata sidecar as a sequence tombstone.
+ *
+ * <p>Legacy windows have no sidecar and are intentionally retained: their
+ * HTTP success is not backed by the transport-window registry, so this class
+ * never guesses that they are safely replayable.
+ */
+public final class AcknowledgedWindowCleaner {
+    private static final Logger log =
+            LoggerFactory.getLogger(AcknowledgedWindowCleaner.class);
+    private static final Pattern PAYLOAD = Pattern.compile(
+            "^([A-Za-z0-9_-]+)\\.([1-9][0-9]*)\\.log\\.gz$");
+
+    private AcknowledgedWindowCleaner() {}
+
+    public static Path acknowledgementPath(
+            Path sessionDir, String channel, long sequence) {
+        return sessionDir.resolve(
+                ".gpufl-window-ack." + channel + "." + sequence);
+    }
+
+    /**
+     * Durable local proof that the backend explicitly acknowledged the
+     * transport identity. A plain 2xx from an older backend is not sufficient.
+     */
+    public static boolean recordBackendAcknowledgement(
+            Path sessionDir, WindowMetadata metadata) {
+        Path marker = acknowledgementPath(
+                sessionDir, metadata.channel(), metadata.windowSequence());
+        if (Files.isRegularFile(marker)) return true;
+        byte[] body = (metadata.windowId() + "\n")
+                .getBytes(StandardCharsets.UTF_8);
+        try (FileChannel out = FileChannel.open(
+                marker, StandardOpenOption.CREATE_NEW,
+                StandardOpenOption.WRITE)) {
+            ByteBuffer bytes = ByteBuffer.wrap(body);
+            while (bytes.hasRemaining()) out.write(bytes);
+            out.force(true);
+            return true;
+        } catch (FileAlreadyExistsException alreadyRecorded) {
+            return Files.isRegularFile(marker);
+        } catch (IOException failure) {
+            log.error("Could not persist backend window acknowledgement {}: {}",
+                    marker, failure.getMessage());
+            return false;
+        }
+    }
+
+    public static boolean hasBackendAcknowledgement(
+            Path sessionDir, String channel, long sequence) {
+        Path metadataPath = sessionDir.resolve(
+                ".gpufl-window." + channel + "." + sequence + ".json");
+        if (!Files.isRegularFile(metadataPath)) return false;
+        try {
+            WindowMetadata metadata = JsonSettings.MAPPER.readValue(
+                    metadataPath.toFile(), WindowMetadata.class);
+            return metadata.channel().equals(channel)
+                    && metadata.windowSequence() == sequence
+                    && hasBackendAcknowledgement(sessionDir, metadata);
+        } catch (RuntimeException malformed) {
+            return false;
+        }
+    }
+
+    private static boolean hasBackendAcknowledgement(
+            Path sessionDir, WindowMetadata metadata) {
+        Path marker = acknowledgementPath(
+                sessionDir, metadata.channel(), metadata.windowSequence());
+        try {
+            return Files.isRegularFile(marker)
+                    && Files.readString(marker).trim()
+                        .equals(metadata.windowId());
+        } catch (IOException unreadable) {
+            return false;
+        }
+    }
+
+    /**
+     * @return true when this was an identity-aware payload and is now absent;
+     *         false when it was legacy/invalid and was deliberately retained.
+     */
+    public static boolean deleteIfIdentityAware(Path payload) {
+        if (payload == null || payload.getFileName() == null
+                || payload.getParent() == null) {
+            return false;
+        }
+        Matcher match = PAYLOAD.matcher(payload.getFileName().toString());
+        if (!match.matches()) return false;
+
+        String channel = match.group(1);
+        long sequence;
+        try {
+            sequence = Long.parseLong(match.group(2));
+        } catch (NumberFormatException malformed) {
+            return false;
+        }
+        Path sessionDir = payload.getParent();
+        Path sessionName = sessionDir.getFileName();
+        if (sessionName == null) return false;
+        Path metadataPath = sessionDir.resolve(
+                ".gpufl-window." + channel + "." + sequence + ".json");
+        if (!Files.isRegularFile(metadataPath)) {
+            return false; // old client: preserve its payload
+        }
+        try {
+            WindowMetadata metadata = JsonSettings.MAPPER.readValue(
+                    metadataPath.toFile(), WindowMetadata.class);
+            if (!metadata.isValidFor(
+                    sessionName.toString(), channel, sequence,
+                    payload.getFileName().toString())) {
+                log.error("Refusing ACK cleanup: metadata {} does not match {}",
+                        metadataPath, payload);
+                return false;
+            }
+            if (!hasBackendAcknowledgement(sessionDir, metadata)) {
+                return false; // old backend: no matching identity proof
+            }
+            Files.deleteIfExists(payload);
+            log.debug("Deleted backend-ACKed payload {}; retained tombstone {}",
+                    payload, metadataPath);
+            return true;
+        } catch (IOException | RuntimeException e) {
+            log.error("Refusing ACK cleanup for {}: {}", payload,
+                    e.getMessage());
+            return false;
+        }
+    }
+}

src/main/java/com/gpuflight/agent/GpuflAgent.java

@@ -24,9 +24,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
 public class GpuflAgent {
@@ -40,6 +42,8 @@ public class GpuflAgent {
     private ExecutorService executor;
     private Publisher publisher;
     private TailerManager tailerManager;
+    private BlockingQueue<Path> acknowledgedWindows;
+    private final AtomicInteger ackCleanupInFlight = new AtomicInteger();
     private final Map<File, List<String>> watchedFolders = new LinkedHashMap<>();
 
     public GpuflAgent(AgentConfig config, String[] args, Map<String, String> env) {
@@ -54,6 +58,8 @@ public void start() throws Exception {
 
         boolean exitWhenDrained = ConfigLoader.parseExitWhenDrained(args, env);
         boolean exitIfEmpty = ConfigLoader.parseExitIfEmpty(args, env);
+        boolean retainAcknowledgedPayloads =
+                ConfigLoader.parseRetainAcknowledgedPayloads(args, env);
         // A launcher-spawned --upload agent uploads only THIS run's sessions. The JVM
         // start time predates the target's session (the launcher spawns the agent
         // before it forks the target), so any session dir older than it is from an
@@ -71,7 +77,7 @@ public void start() throws Exception {
 
         String cursorFile = ConfigLoader.resolve(args, "cursor-file", "GPUFL_CURSOR_FILE", "./cursor.json", env);
         var cursorMgr = new CursorManager(new File(cursorFile));
-        var consumedFilesQueue = new LinkedBlockingQueue<Path>();
+        acknowledgedWindows = new LinkedBlockingQueue<>();
 
         String topicPrefix = topicPrefix(config);
         StreamUploadSettings streamUploadSettings = switch (config.publisher()) {
@@ -86,7 +92,7 @@ public void start() throws Exception {
         executor = Executors.newVirtualThreadPerTaskExecutor();
         var deduplicator = new DeviceMetricDeduplicator();
 
-        tailerManager = new TailerManager(executor, publisher, cursorMgr, consumedFilesQueue,
+        tailerManager = new TailerManager(executor, publisher, cursorMgr, acknowledgedWindows,
                 deduplicator, streamUploadSettings, topicPrefix);
         tailerManager.setPruneFailed(ConfigLoader.parsePruneFailed(args, env));
 
@@ -111,23 +117,11 @@ public void start() throws Exception {
             new SessionWatcher(entry.getKey(), entry.getValue(), spawn).start(executor);
         }
 
-        if (config.archiver() != null) {
-            var archiver = new LogArchiver(config.archiver());
-            executor.submit(() -> {
-                while (!Thread.currentThread().isInterrupted()) {
-                    try {
-                        Path path = consumedFilesQueue.take();
-                        String objectKey = ConfigLoader.buildArchiveKey(config.archiver().prefix(), path);
-                        archiver.archive(path, objectKey);
-                    } catch (InterruptedException e) {
-                        Thread.currentThread().interrupt();
-                        break;
-                    } catch (Exception e) {
-                        System.err.println("[archiver] Error: " + e.getMessage());
-                    }
-                }
-            });
-        }
+        LogArchiver archiver =
+                config.archiver() == null ? null : new LogArchiver(config.archiver());
+        executor.submit(() -> processAcknowledgedWindows(
+                archiver, streamUploadSettings.enabled()
+                        && !retainAcknowledgedPayloads));
 
         Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
 
@@ -178,6 +172,36 @@ private void shutdown() {
         try { if (publisher != null) publisher.close(); } catch (Exception ignored) {}
     }
 
+    private void processAcknowledgedWindows(
+            LogArchiver archiver, boolean identityAckEnabled) {
+        while (!Thread.currentThread().isInterrupted()) {
+            Path path = null;
+            try {
+                path = acknowledgedWindows.take();
+                ackCleanupInFlight.incrementAndGet();
+                if (archiver != null) {
+                    String objectKey = ConfigLoader.buildArchiveKey(
+                            config.archiver().prefix(), path);
+                    archiver.archive(path, objectKey);
+                }
+                if (identityAckEnabled) {
+                    AcknowledgedWindowCleaner.deleteIfIdentityAware(path);
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            } catch (Exception e) {
+                // Preserve the payload. The persisted cursor lets the next
+                // agent start re-enqueue this identity-aware window.
+                System.err.println("[ack-cleanup] Retained "
+                        + (path == null ? "window" : path)
+                        + ": " + e.getMessage());
+            } finally {
+                if (path != null) ackCleanupInFlight.decrementAndGet();
+            }
+        }
+    }
+
     void awaitDrainThenExit(Collection<File> folders, TailerManager tailers, long sinceMs) {
         int clean = 0;
         while (true) {
@@ -186,7 +210,12 @@ void awaitDrainThenExit(Collection<File> folders, TailerManager tailers, long si
             // .tmp/ marker: a short trace finalizes that marker between our 1s polls, which
             // left the old sawActive gate spinning forever even after the upload drained.
             boolean started = tailers.hasStartedAnySession();
-            boolean idle = tailers.getActiveTailers().get() == 0 && !anyActiveSession(folders, sinceMs);
+            boolean cleanupIdle = acknowledgedWindows == null
+                    || (acknowledgedWindows.isEmpty()
+                        && ackCleanupInFlight.get() == 0);
+            boolean idle = tailers.getActiveTailers().get() == 0
+                    && cleanupIdle
+                    && !anyActiveSession(folders, sinceMs);
             clean = (started && idle) ? clean + 1 : 0;
             if (clean >= 2) return;
         }