diff --git a/src/main/java/com/gpuflight/agent/AcknowledgedWindowCleaner.java b/src/main/java/com/gpuflight/agent/AcknowledgedWindowCleaner.java new file mode 100644 index 0000000..d20dce3 --- /dev/null +++ b/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. + * + *

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; + } + } +} diff --git a/src/main/java/com/gpuflight/agent/GpuflAgent.java b/src/main/java/com/gpuflight/agent/GpuflAgent.java index 395099f..27995f8 100644 --- a/src/main/java/com/gpuflight/agent/GpuflAgent.java +++ b/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 acknowledgedWindows; + private final AtomicInteger ackCleanupInFlight = new AtomicInteger(); private final Map> watchedFolders = new LinkedHashMap<>(); public GpuflAgent(AgentConfig config, String[] args, Map 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(); + 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 folders, TailerManager tailers, long sinceMs) { int clean = 0; while (true) { @@ -186,7 +210,12 @@ void awaitDrainThenExit(Collection 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; } diff --git a/src/main/java/com/gpuflight/agent/LogTailer.java b/src/main/java/com/gpuflight/agent/LogTailer.java index 7af0cc8..793c80f 100644 --- a/src/main/java/com/gpuflight/agent/LogTailer.java +++ b/src/main/java/com/gpuflight/agent/LogTailer.java @@ -1,11 +1,14 @@ package com.gpuflight.agent; +import tools.jackson.core.JacksonException; import tools.jackson.databind.JsonNode; import com.gpuflight.agent.config.JsonSettings; import com.gpuflight.agent.config.StreamUploadSettings; import com.gpuflight.agent.filter.DeviceMetricDeduplicator; import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.model.WindowMetadata; import com.gpuflight.agent.publisher.Publisher; +import com.gpuflight.agent.publisher.WindowPublishResult; import com.gpuflight.agent.util.Delays; import java.io.BufferedInputStream; import java.io.ByteArrayOutputStream; @@ -52,6 +55,8 @@ public class LogTailer { /** Bytes of uncompressed head used for the content signature. */ private static final int HEAD_SIG_BYTES = 512; + private static final long PERMANENT_WINDOW_FAILURE = Long.MIN_VALUE; + private String permanentWindowFailureReason; public LogTailer(File folder, String sessionId, String logType, String topicPrefix, CursorManager cursorMgr, BlockingQueue consumedFilesQueue) { @@ -103,6 +108,73 @@ private File resolveRotated(int index) { return gz.exists() ? gz : null; } + private WindowMetadata metadataFor(File window, int index) + throws IOException { + Path metadataPath = metadataPath(index); + if (!Files.exists(metadataPath)) { + SessionOwnership.State ownership = + SessionOwnership.probe(sessionDir().toPath()); + if (ownership == SessionOwnership.State.LEGACY_NO_LOCK) { + return null; // backward-compatible window from an older client + } + // A current client publishes identity before it makes the payload + // visible. Missing metadata is therefore either the tiny recovery + // interval of an older lock-aware build or a damaged contract, not + // permission to silently downgrade exact-once upload to legacy. + throw new IOException( + "identity metadata not visible yet for current-client payload " + + window.getName()); + } + WindowMetadata metadata; + try { + metadata = JsonSettings.MAPPER.readValue( + metadataPath.toFile(), WindowMetadata.class); + } catch (JacksonException malformed) { + throw new PermanentWindowException( + "window identity metadata is malformed for " + + window.getName()); + } + if (!metadata.isValidFor( + sessionId, logType, index, window.getName())) { + throw new PermanentWindowException( + "window metadata does not match payload " + window.getName()); + } + return metadata; + } + + private Path metadataPath(int index) { + return sessionDir().toPath().resolve( + ".gpufl-window." + logType + "." + index + ".json"); + } + + /** + * A crash after persisting the post-ACK cursor but before local cleanup + * must not leak payloads forever. On restart, re-enqueue only windows + * proven identity-aware by their tombstones; legacy cursor history keeps + * its prior retention behavior. + */ + private void enqueuePreviouslyAcknowledgedWindows(int nextIndex) { + if (consumedFilesQueue == null || !streamUploadSettings.enabled()) return; + for (int index = 1; index < nextIndex; ++index) { + File payload = resolveRotated(index); + if (payload != null && isGz(payload) + && Files.isRegularFile(metadataPath(index)) + && AcknowledgedWindowCleaner.hasBackendAcknowledgement( + sessionDir().toPath(), logType, index)) { + consumedFilesQueue.offer(payload.toPath()); + } + } + } + + private static boolean checksumMatches( + WindowMetadata metadata, byte[] payload) { + if (metadata == null) return true; + CRC32 crc = new CRC32(); + crc.update(payload); + return metadata.payloadBytes() == payload.length + && metadata.payloadCrc32() == crc.getValue(); + } + private static boolean isGz(File f) { return f.getName().endsWith(".gz"); } @@ -221,13 +293,38 @@ public boolean tail(Publisher publisher) { // bytes already sent within it (mid-window resume after a crash). int idx = Math.max(1, cursor.fileIndex()); long offset = cursor.fileIndex() >= 1 ? cursor.offset() : 0L; + if (SessionOwnership.hasAgentWindowFailure( + sessionDir().toPath(), logType, idx)) { + System.err.println("[" + logType + "] prior permanent transport " + + "window failure is still present; refusing " + + "session-complete"); + return true; + } + enqueuePreviouslyAcknowledgedWindows(idx); System.out.println("[" + logType + "] Starting (window mode) - window index=" + idx + ", offset=" + offset); while (!Thread.currentThread().isInterrupted()) { File window = resolveRotated(idx); if (window != null) { long resume = drainWindow(window, offset, publisher, idx); - if (resume < 0) { + if (resume == PERMANENT_WINDOW_FAILURE) { + String reason = permanentWindowFailureReason == null + ? "immutable_window_contract_violation" + : permanentWindowFailureReason; + boolean marked = + SessionOwnership.recordAgentWindowFailure( + sessionDir().toPath(), logType, idx, reason); + System.err.println("[" + logType + "] permanent transport " + + "window failure at " + window.getName() + ": " + + reason + ". Payload retained; session-complete " + + "will not be emitted." + + (marked ? "" + : " WARNING: durable loss marker write failed.")); + if (marked) return true; + // Never fall through to session-complete without a + // durable indication that this profile is incomplete. + if (!Delays.sleep(Delays.LOG_TAILER_RETRY)) break; + } else if (resume < 0) { System.out.println("[" + logType + "] Sent window " + window.getName() + "."); if (consumedFilesQueue != null) consumedFilesQueue.offer(window.toPath()); idx++; @@ -241,9 +338,37 @@ public boolean tail(Publisher publisher) { continue; } + // The backend ACK can be durable before local payload cleanup and + // cursor persistence. If cleanup won that race and this agent then + // crashed, the ACK tombstone is enough to skip the absent payload + // and continue with the next sequence on restart. + if (AcknowledgedWindowCleaner.hasBackendAcknowledgement( + sessionDir().toPath(), logType, idx)) { + idx++; + offset = 0L; + cursorMgr.update(streamKey, idx, offset); + continue; + } + // Window not published yet. Is the session still writing? File tmp = sessionTmpDir(); if (tmp.exists()) { + SessionOwnership.State ownership = + SessionOwnership.probe(sessionDir().toPath()); + if (ownership == SessionOwnership.State.ACTIVE + || ownership == SessionOwnership.State.UNKNOWN) { + // A quiet workload can leave `.tmp` unchanged for much + // longer than the legacy mtime grace. The OS lock is the + // authoritative liveness signal and survives that silence. + if (!Delays.sleep(Delays.LOG_TAILER_POLL)) break; + continue; + } + if (ownership == SessionOwnership.State.UNOWNED) { + System.out.println("[" + logType + "] Session finished " + + "(ownership lock released - client gone; last sent " + + (idx - 1) + ")."); + return true; + } // A live client keeps appending to .tmp/ (system sampling alone // advances its mtime); a client that crashed or was killed leaves // .tmp/ frozen and never removed. Once it has sat unchanged past the @@ -261,8 +386,19 @@ public boolean tail(Publisher publisher) { // .tmp/ gone -> the client closed every channel. Each channel's last // window is published BEFORE .tmp/ is removed, so wait a moment then do // one final check for a straggler before finishing. + SessionOwnership.State ownership = + SessionOwnership.probe(sessionDir().toPath()); + if (ownership == SessionOwnership.State.ACTIVE + || ownership == SessionOwnership.State.UNKNOWN) { + if (!Delays.sleep(Delays.LOG_TAILER_POLL)) break; + continue; + } if (!Delays.sleep(Delays.SESSION_END_GRACE_PERIOD)) break; - if (resolveRotated(idx) != null) continue; + if (resolveRotated(idx) != null + || SessionOwnership.probe(sessionDir().toPath()) + == SessionOwnership.State.ACTIVE) { + continue; + } System.out.println("[" + logType + "] Session finished (.tmp gone, no window " + idx + "; last sent " + (idx - 1) + ")."); return false; // clean: producer closed and removed .tmp/ @@ -309,8 +445,28 @@ private long drainWindow(File window, long startOffset, Publisher publisher, int // (.log) windows from a no-compressor build fall through to the line path below. if (streamUploadSettings.enabled() && isGz(window)) { try { + WindowMetadata metadata = metadataFor(window, idx); byte[] gz = Files.readAllBytes(window.toPath()); - if (publisher.publishStreamGz(sessionId, gz)) { + if (!checksumMatches(metadata, gz)) { + permanentWindowFailureReason = + "payload_checksum_mismatch:" + window.getName(); + return PERMANENT_WINDOW_FAILURE; + } + WindowPublishResult published = metadata == null + ? (publisher.publishStreamGz(sessionId, gz) + ? WindowPublishResult.acceptedLegacy() + : WindowPublishResult.retry()) + : publisher.publishTransportWindow( + sessionId, metadata, gz); + if (published.accepted()) { + if (published.identityAcknowledged() + && !AcknowledgedWindowCleaner + .recordBackendAcknowledgement( + sessionDir().toPath(), metadata)) { + // The backend can safely deduplicate the retry. Do not + // advance until deletion authorization is durable. + return 0L; + } // Persist the window-done advance right after the durable 2xx so a crash // before tail() advances won't re-send the whole window on restart (the // line path likewise persists the cursor after each accepted batch). @@ -319,6 +475,9 @@ private long drainWindow(File window, long startOffset, Publisher publisher, int } System.out.println("[" + logType + "] window publish FAILED - retry in 5s"); return 0L; + } catch (PermanentWindowException e) { + permanentWindowFailureReason = e.getMessage(); + return PERMANENT_WINDOW_FAILURE; } catch (IOException e) { System.out.println("[" + logType + "] read error on " + window.getName() + ": " + e.getMessage()); return 0L; @@ -508,4 +667,10 @@ private static long skipFully(InputStream in, long n) throws IOException { } return total; } + + private static final class PermanentWindowException extends IOException { + PermanentWindowException(String message) { + super(message); + } + } } diff --git a/src/main/java/com/gpuflight/agent/SessionOwnership.java b/src/main/java/com/gpuflight/agent/SessionOwnership.java new file mode 100644 index 0000000..efdee95 --- /dev/null +++ b/src/main/java/com/gpuflight/agent/SessionOwnership.java @@ -0,0 +1,119 @@ +package com.gpuflight.agent; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.channels.OverlappingFileLockException; +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.stream.Stream; + +/** + * Cross-process session liveness contract shared with gpufl-client. + * + *

The client holds an exclusive OS lock on {@value #LOCK_FILE} for the + * complete writer lifetime. Finished root windows remain uploadable while it + * is held; only orphan handling and session completion depend on this probe. + */ +public final class SessionOwnership { + public static final String LOCK_FILE = ".gpufl-session.lock"; + public static final String LOSS_PREFIX = ".gpufl-transport-loss."; + + public enum State { + /** Another live process owns the session. */ + ACTIVE, + /** The lock file exists and its OS lock is currently acquirable. */ + UNOWNED, + /** Produced by a client older than the ownership-lock contract. */ + LEGACY_NO_LOCK, + /** Conservatively treated as active; never finalize through an I/O error. */ + UNKNOWN + } + + private SessionOwnership() {} + + public static State probe(Path sessionDir) { + Path path = sessionDir.resolve(LOCK_FILE); + if (!Files.exists(path)) { + return State.LEGACY_NO_LOCK; + } + try (FileChannel channel = FileChannel.open( + path, StandardOpenOption.READ, StandardOpenOption.WRITE)) { + try (FileLock ignored = channel.tryLock()) { + return ignored == null ? State.ACTIVE : State.UNOWNED; + } catch (OverlappingFileLockException activeInThisJvm) { + return State.ACTIVE; + } + } catch (IOException | RuntimeException unreadable) { + return State.UNKNOWN; + } + } + + public static boolean hasTransportLoss(Path sessionDir) { + try (Stream entries = Files.list(sessionDir)) { + return entries.anyMatch(path -> + path.getFileName().toString().startsWith(LOSS_PREFIX)); + } catch (IOException e) { + // Completion through an unreadable directory is unsafe. + return true; + } + } + + public static Path agentWindowFailureMarker( + Path sessionDir, String channel, int sequence) { + String safeChannel = channel == null + ? "unknown" + : channel.replaceAll("[^A-Za-z0-9_-]", "_"); + return sessionDir.resolve( + LOSS_PREFIX + "agent-" + safeChannel + "." + sequence + ".json"); + } + + public static boolean hasAgentWindowFailure( + Path sessionDir, String channel, int sequence) { + return Files.isRegularFile( + agentWindowFailureMarker(sessionDir, channel, sequence)); + } + + /** + * Persist a terminal local identity failure. The payload and sidecar stay + * in place for forensic inspection; the shared loss prefix prevents the + * session from being reported complete. + */ + public static boolean recordAgentWindowFailure( + Path sessionDir, String channel, int sequence, String reason) { + Path marker = agentWindowFailureMarker(sessionDir, channel, sequence); + if (Files.isRegularFile(marker)) return true; + String json = "{\"schema_version\":1," + + "\"source\":\"gpufl-agent\"," + + "\"channel\":\"" + jsonEscape(channel) + "\"," + + "\"window_sequence\":" + sequence + "," + + "\"reason\":\"" + jsonEscape(reason) + "\"}\n"; + try { + Files.createDirectories(sessionDir); + try (FileChannel out = FileChannel.open( + marker, StandardOpenOption.CREATE_NEW, + StandardOpenOption.WRITE)) { + ByteBuffer bytes = StandardCharsets.UTF_8.encode(json); + while (bytes.hasRemaining()) out.write(bytes); + out.force(true); + } + return true; + } catch (FileAlreadyExistsException alreadyRecorded) { + return Files.isRegularFile(marker); + } catch (IOException e) { + return false; + } + } + + private static String jsonEscape(String value) { + if (value == null) return ""; + return value.replace("\\", "\\\\") + .replace("\"", "\\\"") + .replace("\r", "\\r") + .replace("\n", "\\n"); + } +} diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java index 0c3433d..3bbc1db 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -149,6 +149,20 @@ public static boolean parseExitWhenDrained(String[] args, Map en return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); } + /** + * Keep identity-aware payload files after backend acknowledgement. The + * experiment supervisor uses this because it computes quick results from + * those same files after its one-shot uploader exits, then removes the + * entire run directory itself. + */ + public static boolean parseRetainAcknowledgedPayloads( + String[] args, Map env) { + String v = resolve( + args, "retain-acked-payloads", + "GPUFL_AGENT_RETAIN_ACKED_PAYLOADS", null, env); + return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); + } + /** * Scope a launcher-spawned {@code --upload} agent to THIS run: ignore every * session that already exists when the agent starts, uploading only sessions @@ -239,6 +253,7 @@ Config file (overrides all flags): --upload-mode= Upload protocol [GPUFL_AGENT_UPLOAD_MODE] default: stream --stream-max-lines= Stream batch line limit [GPUFL_AGENT_STREAM_MAX_LINES] default: 5000 --stream-max-bytes= Stream batch byte limit [GPUFL_AGENT_STREAM_MAX_BYTES] default: 1000000 + --retain-acked-payloads= Keep local window payloads after ACK [GPUFL_AGENT_RETAIN_ACKED_PAYLOADS] default: false Kafka publisher: --brokers= Bootstrap servers [GPUFL_KAFKA_BROKERS] diff --git a/src/main/java/com/gpuflight/agent/model/WindowMetadata.java b/src/main/java/com/gpuflight/agent/model/WindowMetadata.java new file mode 100644 index 0000000..990545d --- /dev/null +++ b/src/main/java/com/gpuflight/agent/model/WindowMetadata.java @@ -0,0 +1,32 @@ +package com.gpuflight.agent.model; + +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Immutable client sidecar for one transport window. */ +public record WindowMetadata( + @JsonProperty("schema_version") int schemaVersion, + @JsonProperty("type") String type, + @JsonProperty("window_id") String windowId, + @JsonProperty("session_id") String sessionId, + @JsonProperty("channel") String channel, + @JsonProperty("window_sequence") long windowSequence, + @JsonProperty("opened_mono_ms") long openedMonoMs, + @JsonProperty("closed_mono_ms") long closedMonoMs, + @JsonProperty("created_wall_ms") long createdWallMs, + @JsonProperty("payload_file") String payloadFile, + @JsonProperty("payload_bytes") long payloadBytes, + @JsonProperty("payload_crc32") long payloadCrc32 +) { + public boolean isValidFor( + String expectedSession, String expectedChannel, + long expectedSequence, String expectedPayload) { + return schemaVersion == 1 + && "transport_window".equals(type) + && windowId != null && !windowId.isBlank() + && expectedSession.equals(sessionId) + && expectedChannel.equals(channel) + && expectedSequence == windowSequence + && expectedPayload.equals(payloadFile) + && payloadBytes >= 0; + } +} diff --git a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java index 48a320e..13a1459 100644 --- a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java +++ b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java @@ -3,6 +3,7 @@ import com.gpuflight.agent.config.HttpConfig; import com.gpuflight.agent.config.JsonSettings; import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.model.WindowMetadata; import java.io.ByteArrayOutputStream; import java.net.InetAddress; @@ -103,12 +104,42 @@ public boolean publishStreamGz(String sessionId, byte[] gzBody) { return postGzStream(sessionId, gzBody); } + @Override + public boolean publishStreamGz( + String sessionId, WindowMetadata window, byte[] gzBody) { + return publishTransportWindow(sessionId, window, gzBody).accepted(); + } + + @Override + public WindowPublishResult publishTransportWindow( + String sessionId, WindowMetadata window, byte[] gzBody) { + if (window == null) { + return publishStreamGz(sessionId, gzBody) + ? WindowPublishResult.acceptedLegacy() + : WindowPublishResult.retry(); + } + if (gzBody == null || gzBody.length == 0) { + return WindowPublishResult.acceptedLegacy(); + } + System.out.println("[agent] HTTP stream POST starting (window): url=" + + config.streamEndpoint() + " session=" + sessionId + + " window=" + window.windowId() + + " sequence=" + window.windowSequence() + + " gzipBytes=" + gzBody.length); + return postGzStreamResult(sessionId, gzBody, window); + } + /** * POST an already-gzipped NDJSON body to the stream endpoint and interpret the * status. 2xx (accepted), 409 (already finalized on a restart) and 402 (limit) * all advance; anything else is a retryable failure. */ private boolean postGzStream(String sessionId, byte[] gzBody) { + return postGzStreamResult(sessionId, gzBody, null).accepted(); + } + + private WindowPublishResult postGzStreamResult( + String sessionId, byte[] gzBody, WindowMetadata window) { try { String url = config.streamEndpoint(); var requestBuilder = HttpRequest.newBuilder() @@ -118,6 +149,15 @@ private boolean postGzStream(String sessionId, byte[] gzBody) { .header("X-GpuFlight-Session-Id", sessionId) .header("X-GpuFlight-Hostname", InetAddress.getLocalHost().getHostName()) .POST(HttpRequest.BodyPublishers.ofByteArray(gzBody)); + if (window != null) { + requestBuilder + .header("X-GpuFlight-Window-Id", window.windowId()) + .header("X-GpuFlight-Window-Channel", window.channel()) + .header("X-GpuFlight-Window-Sequence", + Long.toString(window.windowSequence())) + .header("X-GpuFlight-Window-CRC32", + Long.toUnsignedString(window.payloadCrc32())); + } addAuthHeader(requestBuilder); @@ -125,7 +165,26 @@ private boolean postGzStream(String sessionId, byte[] gzBody) { int sc = response.statusCode(); if (sc >= 200 && sc <= 299) { System.out.println("[agent] HTTP stream POST accepted: status=" + sc + " session=" + sessionId); - return true; + if (window != null && identityAcknowledged(response.body())) { + return WindowPublishResult.acceptedIdentity(); + } + if (window != null) { + System.err.println("[GPUFL] backend accepted transport " + + "window through a legacy/unconfirmed path; local " + + "payload will be retained, session=" + sessionId + + " window=" + window.windowId()); + } + return WindowPublishResult.acceptedLegacy(); + } + if (sc == 409 && window != null + && response.body().contains("window_identity_conflict")) { + // Reusing an immutable UUID for different bytes is corruption, + // not the legacy "session already finalized" terminal success. + // Do not advance the cursor or delete the payload. + System.err.println("[GPUFL] transport window identity conflict - " + + "refusing acknowledgement, session=" + sessionId + + " window=" + window.windowId()); + return WindowPublishResult.retry(); } if (sc == 409) { // Session already finalized on the backend - typically this agent @@ -135,21 +194,31 @@ private boolean postGzStream(String sessionId, byte[] gzBody) { // re-POSTing forever. System.out.println("[agent] HTTP stream POST: session already uploaded (409) - " + "skipping, session=" + sessionId); - return true; + return WindowPublishResult.acceptedLegacy(); } if (sc == 402) { // GPU/workspace limit exceeded - permanent; retrying won't help. System.err.println("[GPUFL] GPU limit exceeded for this workspace - " + "skipping session=" + sessionId + ". " + response.body()); - return true; + return WindowPublishResult.acceptedLegacy(); } System.out.println("HTTP stream publish failed [" + url + "]: " + sc + " - " + response.body()); - return false; + return WindowPublishResult.retry(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - return false; + return WindowPublishResult.retry(); } catch (Exception e) { logConnectionError("HTTP stream", e); + return WindowPublishResult.retry(); + } + } + + private static boolean identityAcknowledged(String body) { + if (body == null || body.isBlank()) return false; + try { + return JsonSettings.MAPPER.readTree(body) + .path("window_acknowledged").asBoolean(false); + } catch (Exception malformedResponse) { return false; } } diff --git a/src/main/java/com/gpuflight/agent/publisher/Publisher.java b/src/main/java/com/gpuflight/agent/publisher/Publisher.java index b329242..8e3fd01 100644 --- a/src/main/java/com/gpuflight/agent/publisher/Publisher.java +++ b/src/main/java/com/gpuflight/agent/publisher/Publisher.java @@ -1,6 +1,7 @@ package com.gpuflight.agent.publisher; import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.model.WindowMetadata; import java.io.Closeable; import java.util.List; @@ -25,6 +26,27 @@ default boolean publishStreamGz(String sessionId, byte[] gzBody) { return false; } + /** + * Identity-aware variant. Implementations that have not adopted the + * window contract retain the legacy behavior through this default. + */ + default boolean publishStreamGz( + String sessionId, WindowMetadata window, byte[] gzBody) { + return publishStreamGz(sessionId, gzBody); + } + + /** + * Identity-aware result used by the tailer. Legacy implementations can + * accept the bytes, but may not authorize local deletion because they + * cannot prove that a transport-window identity was registered. + */ + default WindowPublishResult publishTransportWindow( + String sessionId, WindowMetadata window, byte[] gzBody) { + return publishStreamGz(sessionId, window, gzBody) + ? WindowPublishResult.acceptedLegacy() + : WindowPublishResult.retry(); + } + /** * Signal the backend that EVERY channel of {@code sessionId} has finished * uploading — the agent has drained all per-channel tailers and every batch diff --git a/src/main/java/com/gpuflight/agent/publisher/WindowPublishResult.java b/src/main/java/com/gpuflight/agent/publisher/WindowPublishResult.java new file mode 100644 index 0000000..8e6d826 --- /dev/null +++ b/src/main/java/com/gpuflight/agent/publisher/WindowPublishResult.java @@ -0,0 +1,25 @@ +package com.gpuflight.agent.publisher; + +/** + * Separates backend acceptance from authorization to delete the local payload. + * + *

An older backend can accept an identity-aware request through its legacy + * path. That is enough to advance the upload cursor, but not enough to delete + * the only replayable copy because no transport-window registry row exists. + */ +public record WindowPublishResult( + boolean accepted, + boolean identityAcknowledged) { + + public static WindowPublishResult retry() { + return new WindowPublishResult(false, false); + } + + public static WindowPublishResult acceptedLegacy() { + return new WindowPublishResult(true, false); + } + + public static WindowPublishResult acceptedIdentity() { + return new WindowPublishResult(true, true); + } +} diff --git a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java index 6e7dea9..342443a 100644 --- a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -22,6 +22,9 @@ public class SessionWatcher { private static final Pattern CHANNEL_FILE_PATTERN = Pattern.compile("^(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$"); + private static final Pattern WINDOW_TOMBSTONE_PATTERN = + Pattern.compile("^\\.gpufl-window\\." + + "(device|scope|system|sass)\\.[1-9][0-9]*\\.json$"); /** Marker file the agent drops in a session dir once every channel has been * fully uploaded. Discovery skips a session that has it, so a re-scan never @@ -132,7 +135,9 @@ private static boolean looksLikeSession(File dir) { if (inside == null) return false; for (File child : inside) { if (child.isFile() && - CHANNEL_FILE_PATTERN.matcher(child.getName()).matches()) { + (CHANNEL_FILE_PATTERN.matcher(child.getName()).matches() + || WINDOW_TOMBSTONE_PATTERN.matcher( + child.getName()).matches())) { return true; } } diff --git a/src/main/java/com/gpuflight/agent/service/TailerManager.java b/src/main/java/com/gpuflight/agent/service/TailerManager.java index e9a6430..372b962 100644 --- a/src/main/java/com/gpuflight/agent/service/TailerManager.java +++ b/src/main/java/com/gpuflight/agent/service/TailerManager.java @@ -2,6 +2,7 @@ import com.gpuflight.agent.CursorManager; import com.gpuflight.agent.LogTailer; +import com.gpuflight.agent.SessionOwnership; import com.gpuflight.agent.config.StreamUploadSettings; import com.gpuflight.agent.filter.DeviceMetricDeduplicator; import com.gpuflight.agent.model.DiscoveredSession; @@ -74,9 +75,16 @@ public void spawnSessionTailers(DiscoveredSession session) { if (tailer.tail(publisher)) orphaned.set(true); // finished off a stale .tmp/ if (!Thread.currentThread().isInterrupted() && remaining.decrementAndGet() == 0) { + boolean transportLoss = + SessionOwnership.hasTransportLoss(sessionDir.toPath()); + if (transportLoss) { + orphaned.set(true); + log.error("session {} has a durable transport-loss " + + "marker; refusing session-complete", sid); + } // Last channel done: every available window is uploaded. Signal the // backend (stream mode), then settle the session so a re-scan skips it. - if (streamUploadSettings.enabled()) { + if (streamUploadSettings.enabled() && !transportLoss) { signalSessionComplete(publisher, sid); } settleSession(sessionDir, orphaned.get()); diff --git a/src/test/java/com/gpuflight/agent/AcknowledgedWindowCleanerTest.java b/src/test/java/com/gpuflight/agent/AcknowledgedWindowCleanerTest.java new file mode 100644 index 0000000..0cc1b81 --- /dev/null +++ b/src/test/java/com/gpuflight/agent/AcknowledgedWindowCleanerTest.java @@ -0,0 +1,123 @@ +package com.gpuflight.agent; + +import com.gpuflight.agent.config.JsonSettings; +import com.gpuflight.agent.model.WindowMetadata; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.jupiter.api.Assertions.*; + +class AcknowledgedWindowCleanerTest { + + @Test + void ackDeletesIdentityPayloadButRetainsSequenceTombstone( + @TempDir Path root) throws Exception { + Path session = Files.createDirectories(root.resolve("session-a")); + Path payload = session.resolve("device.1.log.gz"); + Files.write(payload, new byte[] {1, 2, 3}); + Path metadata = writeMetadata( + session, "session-a", "device", 1, payload.getFileName().toString()); + recordAck(session, metadata); + + assertTrue(AcknowledgedWindowCleaner.deleteIfIdentityAware(payload)); + assertFalse(Files.exists(payload)); + assertTrue(Files.isRegularFile(metadata)); + } + + @Test + void identitySidecarWithoutBackendAckCannotAuthorizeDeletion( + @TempDir Path root) throws Exception { + Path session = Files.createDirectories(root.resolve("session-a")); + Path payload = session.resolve("device.1.log.gz"); + Files.write(payload, new byte[] {1, 2, 3}); + writeMetadata( + session, "session-a", "device", 1, + payload.getFileName().toString()); + + assertFalse(AcknowledgedWindowCleaner.deleteIfIdentityAware(payload)); + assertTrue(Files.isRegularFile(payload)); + } + + @Test + void ackForDifferentWindowIdCannotAuthorizeDeletion( + @TempDir Path root) throws Exception { + Path session = Files.createDirectories(root.resolve("session-a")); + Path payload = session.resolve("device.1.log.gz"); + Files.write(payload, new byte[] {1, 2, 3}); + writeMetadata( + session, "session-a", "device", 1, + payload.getFileName().toString()); + Files.writeString( + AcknowledgedWindowCleaner.acknowledgementPath( + session, "device", 1), + "different-window-id\n"); + + assertFalse(AcknowledgedWindowCleaner.deleteIfIdentityAware(payload)); + assertTrue(Files.isRegularFile(payload)); + } + + @Test + void legacyPayloadWithoutMetadataIsRetained(@TempDir Path root) + throws Exception { + Path session = Files.createDirectories(root.resolve("legacy")); + Path payload = session.resolve("device.1.log.gz"); + Files.write(payload, new byte[] {1}); + + assertFalse(AcknowledgedWindowCleaner.deleteIfIdentityAware(payload)); + assertTrue(Files.isRegularFile(payload)); + } + + @Test + void plainWindowIsNotMistakenForIdentityAwareGzip( + @TempDir Path root) throws Exception { + Path session = Files.createDirectories(root.resolve("session-a")); + Path payload = session.resolve("device.1.log"); + Files.write(payload, new byte[] {1}); + writeMetadata( + session, "session-a", "device", 1, + payload.getFileName().toString()); + + assertFalse(AcknowledgedWindowCleaner.deleteIfIdentityAware(payload)); + assertTrue(Files.isRegularFile(payload)); + } + + @Test + void mismatchedMetadataCannotAuthorizeDeletion(@TempDir Path root) + throws Exception { + Path session = Files.createDirectories(root.resolve("session-a")); + Path payload = session.resolve("device.1.log.gz"); + Files.write(payload, new byte[] {1}); + Path metadata = writeMetadata( + session, "different-session", "device", 1, + payload.getFileName().toString()); + recordAck(session, metadata); + + assertFalse(AcknowledgedWindowCleaner.deleteIfIdentityAware(payload)); + assertTrue(Files.isRegularFile(payload)); + } + + private static Path writeMetadata( + Path sessionDir, String sessionId, String channel, + long sequence, String payloadFile) throws Exception { + WindowMetadata metadata = new WindowMetadata( + 1, "transport_window", + "11111111-2222-4333-8444-555555555555", + sessionId, channel, sequence, 10, 20, 30, + payloadFile, 3, 123); + Path path = sessionDir.resolve( + ".gpufl-window." + channel + "." + sequence + ".json"); + Files.writeString(path, JsonSettings.MAPPER.writeValueAsString(metadata)); + return path; + } + + private static void recordAck(Path session, Path metadata) + throws Exception { + WindowMetadata value = JsonSettings.MAPPER.readValue( + metadata.toFile(), WindowMetadata.class); + assertTrue(AcknowledgedWindowCleaner.recordBackendAcknowledgement( + session, value)); + } +} diff --git a/src/test/java/com/gpuflight/agent/LogTailerTest.java b/src/test/java/com/gpuflight/agent/LogTailerTest.java index 38cb9c9..7c61d0e 100644 --- a/src/test/java/com/gpuflight/agent/LogTailerTest.java +++ b/src/test/java/com/gpuflight/agent/LogTailerTest.java @@ -2,7 +2,9 @@ import com.gpuflight.agent.config.StreamUploadSettings; import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.model.WindowMetadata; import com.gpuflight.agent.publisher.Publisher; +import com.gpuflight.agent.publisher.WindowPublishResult; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -12,6 +14,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -37,6 +40,7 @@ static class CapturingPublisher implements Publisher { static class CapturingStreamPublisher implements Publisher { final List> batches = new CopyOnWriteArrayList<>(); final List sessionIds = new CopyOnWriteArrayList<>(); + final List windows = new CopyOnWriteArrayList<>(); @Override public boolean publish(String topic, String key, LogWrapper log) { return false; } @Override public boolean publishStream(String sessionId, List ndjsonLines) { sessionIds.add(sessionId); batches.add(List.copyOf(ndjsonLines)); return true; @@ -50,9 +54,40 @@ static class CapturingStreamPublisher implements Publisher { } catch (Exception e) { return false; } return true; } + @Override public boolean publishStreamGz( + String sessionId, WindowMetadata window, byte[] gzBody) { + windows.add(window); + return publishStreamGz(sessionId, gzBody); + } + @Override public WindowPublishResult publishTransportWindow( + String sessionId, WindowMetadata window, byte[] gzBody) { + windows.add(window); + return publishStreamGz(sessionId, gzBody) + ? WindowPublishResult.acceptedIdentity() + : WindowPublishResult.retry(); + } @Override public void close() {} } + static class FailingStreamPublisher extends CapturingStreamPublisher { + final CountDownLatch attempted = new CountDownLatch(1); + + @Override public WindowPublishResult publishTransportWindow( + String sessionId, WindowMetadata window, byte[] gzBody) { + attempted.countDown(); + return WindowPublishResult.retry(); + } + } + + static class LegacyAcceptingPublisher extends CapturingStreamPublisher { + @Override public WindowPublishResult publishTransportWindow( + String sessionId, WindowMetadata window, byte[] gzBody) { + windows.add(window); + publishStreamGz(sessionId, gzBody); + return WindowPublishResult.acceptedLegacy(); + } + } + private static void gzip(Path src, Path dest) throws IOException { try (var in = Files.newInputStream(src); var out = new GZIPOutputStream(Files.newOutputStream(dest))) { in.transferTo(out); @@ -76,6 +111,25 @@ private static Path gzWindow(Path dir, String session, String channel, int index return gz; } + private static WindowMetadata metadata( + Path dir, String session, String channel, int index, Path payload) + throws IOException { + byte[] bytes = Files.readAllBytes(payload); + java.util.zip.CRC32 crc = new java.util.zip.CRC32(); + crc.update(bytes); + WindowMetadata metadata = new WindowMetadata( + 1, "transport_window", + "11111111-2222-4333-8444-555555555555", + session, channel, index, 10, 20, 30, + payload.getFileName().toString(), bytes.length, crc.getValue()); + Files.writeString( + dir.resolve(session + "/.gpufl-window." + channel + "." + + index + ".json"), + com.gpuflight.agent.config.JsonSettings.MAPPER + .writeValueAsString(metadata)); + return metadata; + } + private static Thread startTailer(LogTailer tailer, Publisher publisher) { Thread t = new Thread(() -> tailer.tail(publisher)); t.setDaemon(true); @@ -99,6 +153,153 @@ private static LogTailer streamTailer(Path dir, String session, String channel, // ---- a complete window is sent whole ---- + @Test + void restartRequeuesIdentityWindowAcknowledgedBeforeCleanup( + @TempDir Path dir) throws Exception { + Path payload = gzWindow( + dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"); + WindowMetadata acknowledged = + metadata(dir, "app", "device", 1, payload); + assertTrue(AcknowledgedWindowCleaner.recordBackendAcknowledgement( + dir.resolve("app"), acknowledged)); + Path cursorFile = dir.resolve("cursor.json"); + CursorManager cursors = new CursorManager(cursorFile.toFile()); + cursors.update("app.device", 2, 0L); + var cleanup = new LinkedBlockingQueue(); + var tailer = new LogTailer( + dir.toFile(), "app", "device", "gpu-trace", + cursors, cleanup, null, + new StreamUploadSettings(true, 10, 1_000_000L)); + + Thread thread = startTailer(tailer, new CapturingPublisher()); + Path requeued = cleanup.poll(2, TimeUnit.SECONDS); + thread.interrupt(); + thread.join(2000); + + assertEquals(payload, requeued); + } + + @Test + void oldBackendAcceptanceAdvancesButRetainsPayload( + @TempDir Path dir) throws Exception { + Path payload = gzWindow( + dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"); + metadata(dir, "app", "device", 1, payload); + Path cursorFile = dir.resolve("cursor.json"); + var cleanup = new LinkedBlockingQueue(); + var publisher = new LegacyAcceptingPublisher(); + var tailer = new LogTailer( + dir.toFile(), "app", "device", "gpu-trace", + new CursorManager(cursorFile.toFile()), cleanup, null, + new StreamUploadSettings(true, 10, 1_000_000L)); + var thread = startTailer(tailer, publisher); + + Path offered = cleanup.poll(2, TimeUnit.SECONDS); + thread.interrupt(); + thread.join(2000); + + assertEquals(payload, offered); + assertFalse(AcknowledgedWindowCleaner.deleteIfIdentityAware(offered)); + assertTrue(Files.exists(payload)); + assertEquals(2, new CursorManager(cursorFile.toFile()) + .get("app.device").fileIndex()); + assertFalse(AcknowledgedWindowCleaner.hasBackendAcknowledgement( + dir.resolve("app"), "device", 1)); + } + + @Test + void restartAfterAckCleanupSkipsTombstoneAndUploadsNextWindow( + @TempDir Path dir) throws Exception { + Path first = gzWindow( + dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"); + metadata(dir, "app", "device", 1, first); + Path cursorFile = dir.resolve("cursor.json"); + var cleanup = new LinkedBlockingQueue(); + var publisher = new CapturingStreamPublisher(); + var firstRun = new LogTailer( + dir.toFile(), "app", "device", "gpu-trace", + new CursorManager(cursorFile.toFile()), cleanup, null, + new StreamUploadSettings(true, 10, 1_000_000L)); + var firstThread = startTailer(firstRun, publisher); + + Path acknowledged = cleanup.poll(2, TimeUnit.SECONDS); + assertEquals(first, acknowledged); + assertTrue(AcknowledgedWindowCleaner.deleteIfIdentityAware(first)); + firstThread.interrupt(); + firstThread.join(2000); + + // Recreate the crash-sensitive state: ACK+cleanup is durable, but the + // cursor update did not survive. The restart must skip sequence 1 via + // the ACK tombstone and continue with sequence 2. + new CursorManager(cursorFile.toFile()).update("app.device", 1, 0L); + Path second = gzWindow( + dir, "app", "device", 2, + "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"); + WindowMetadata secondMetadata = + metadata(dir, "app", "device", 2, second); + var recovered = new CapturingStreamPublisher(); + var secondThread = startTailer( + streamTailer( + dir, "app", "device", + new CursorManager(cursorFile.toFile())), + recovered); + awaitEvents(recovered.windows, 1, 3000); + secondThread.interrupt(); + secondThread.join(2000); + + assertEquals(1, recovered.windows.size()); + assertEquals(secondMetadata.windowId(), + recovered.windows.get(0).windowId()); + } + + @Test + void outageThenAgentRestartRetainsAndRetriesTheSameWindow( + @TempDir Path dir) throws Exception { + Path payload = gzWindow( + dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"); + WindowMetadata expected = + metadata(dir, "app", "device", 1, payload); + Path cursorFile = dir.resolve("cursor.json"); + var firstCursors = new CursorManager(cursorFile.toFile()); + var outage = new FailingStreamPublisher(); + var first = startTailer( + streamTailer(dir, "app", "device", firstCursors), outage); + + assertTrue(outage.attempted.await(2, TimeUnit.SECONDS)); + first.interrupt(); + first.join(2000); + + assertEquals(0, new CursorManager(cursorFile.toFile()) + .get("app.device").fileIndex()); + assertTrue(Files.exists(payload), + "an unacknowledged window must survive agent shutdown"); + + var recovered = new CapturingStreamPublisher(); + var second = startTailer( + streamTailer( + dir, "app", "device", + new CursorManager(cursorFile.toFile())), + recovered); + awaitEvents(recovered.windows, 1, 3000); + long deadline = System.currentTimeMillis() + 2000; + while (new CursorManager(cursorFile.toFile()) + .get("app.device").fileIndex() < 2 + && System.currentTimeMillis() < deadline) { + Thread.sleep(20); + } + second.interrupt(); + second.join(2000); + + assertEquals(1, recovered.windows.size()); + assertEquals(expected.windowId(), recovered.windows.get(0).windowId()); + assertEquals(2, new CursorManager(cursorFile.toFile()) + .get("app.device").fileIndex()); + } + @Test void window_readsAndPublishes(@TempDir Path dir) throws Exception { window(dir, "app", "device", 1, @@ -301,4 +502,139 @@ void sentWindow_offeredToQueue(@TempDir Path dir) throws Exception { assertNotNull(consumed, "the sent window should be offered to the archive queue"); assertEquals(gz.toAbsolutePath(), consumed.toAbsolutePath()); } + + @Test + void identitySidecarIsVerifiedAndForwarded(@TempDir Path dir) + throws Exception { + Path gz = gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\"}\n"); + WindowMetadata expected = metadata( + dir, "app", "device", 1, gz); + var publisher = new CapturingStreamPublisher(); + var thread = startTailer( + streamTailer(dir, "app", "device", + new CursorManager(dir.resolve("cursor.json").toFile())), + publisher); + awaitEvents(publisher.batches, 1, 5000); + thread.interrupt(); + thread.join(2000); + + assertEquals(1, publisher.windows.size()); + assertEquals(expected.windowId(), + publisher.windows.get(0).windowId()); + } + + @Test + void checksumMismatchNeverUploadsTheWindow(@TempDir Path dir) + throws Exception { + Path gz = gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\"}\n"); + metadata(dir, "app", "device", 1, gz); + Files.write(gz, new byte[] {1, 2, 3, 4}); + + var cursors = new CursorManager(dir.resolve("cursor.json").toFile()); + var publisher = new CapturingStreamPublisher(); + var thread = startTailer( + streamTailer(dir, "app", "device", + cursors), + publisher); + thread.join(2000); + + assertFalse(thread.isAlive(), + "immutable checksum mismatch must not retry forever"); + assertTrue(publisher.batches.isEmpty()); + assertTrue(publisher.windows.isEmpty()); + assertTrue(SessionOwnership.hasAgentWindowFailure( + dir.resolve("app"), "device", 1)); + assertEquals(0, cursors.get("app.device").fileIndex(), + "terminal failure must not advance the upload cursor"); + } + + @Test + void currentClientWindowWithoutMetadataNeverDowngradesToLegacy( + @TempDir Path dir) throws Exception { + gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\"}\n"); + Files.createFile( + dir.resolve("app").resolve(SessionOwnership.LOCK_FILE)); + + var publisher = new CapturingStreamPublisher(); + var thread = startTailer( + streamTailer(dir, "app", "device", + new CursorManager(dir.resolve("cursor.json").toFile())), + publisher); + Thread.sleep(300); + thread.interrupt(); + thread.join(2000); + + assertTrue(publisher.batches.isEmpty(), + "a lock-aware session must wait for identity metadata"); + assertTrue(publisher.windows.isEmpty()); + } + + @Test + void mismatchedMetadataMarksTheSessionFailedWithoutUploading( + @TempDir Path dir) throws Exception { + Path gz = gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\"}\n"); + WindowMetadata wrong = new WindowMetadata( + 1, "transport_window", + "11111111-2222-4333-8444-555555555555", + "another-session", "device", 1, 10, 20, 30, + gz.getFileName().toString(), Files.size(gz), 0); + Files.writeString( + dir.resolve("app/.gpufl-window.device.1.json"), + com.gpuflight.agent.config.JsonSettings.MAPPER + .writeValueAsString(wrong)); + + var publisher = new CapturingStreamPublisher(); + var thread = startTailer( + streamTailer(dir, "app", "device", + new CursorManager(dir.resolve("cursor.json").toFile())), + publisher); + thread.join(2000); + + assertFalse(thread.isAlive()); + assertTrue(publisher.windows.isEmpty()); + assertTrue(SessionOwnership.hasAgentWindowFailure( + dir.resolve("app"), "device", 1)); + assertTrue(Files.exists(gz), "forensic payload must be retained"); + } + + @Test + void malformedMetadataMarksTheSessionFailedWithoutUploading( + @TempDir Path dir) throws Exception { + Path gz = gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\"}\n"); + Files.writeString( + dir.resolve("app/.gpufl-window.device.1.json"), + "{not-json"); + + var publisher = new CapturingStreamPublisher(); + var thread = startTailer( + streamTailer(dir, "app", "device", + new CursorManager(dir.resolve("cursor.json").toFile())), + publisher); + thread.join(2000); + + assertFalse(thread.isAlive()); + assertTrue(publisher.windows.isEmpty()); + assertTrue(SessionOwnership.hasAgentWindowFailure( + dir.resolve("app"), "device", 1)); + assertTrue(Files.exists(gz)); + } + + @Test + void durableWindowFailureStopsRetryingAfterAgentRestart( + @TempDir Path dir) throws Exception { + Path session = dir.resolve("app"); + Files.createDirectories(session); + assertTrue(SessionOwnership.recordAgentWindowFailure( + session, "device", 1, "payload_checksum_mismatch")); + var tailer = streamTailer( + dir, "app", "device", + new CursorManager(dir.resolve("cursor.json").toFile())); + + assertTrue(tailer.tail(new CapturingStreamPublisher())); + } } diff --git a/src/test/java/com/gpuflight/agent/SessionOwnershipTest.java b/src/test/java/com/gpuflight/agent/SessionOwnershipTest.java new file mode 100644 index 0000000..60c541a --- /dev/null +++ b/src/test/java/com/gpuflight/agent/SessionOwnershipTest.java @@ -0,0 +1,45 @@ +package com.gpuflight.agent; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; + +import static org.junit.jupiter.api.Assertions.*; + +class SessionOwnershipTest { + + @Test + void osLockOverridesQuietDirectoryHeuristics(@TempDir Path session) throws Exception { + assertEquals(SessionOwnership.State.LEGACY_NO_LOCK, + SessionOwnership.probe(session)); + + Path lockPath = session.resolve(SessionOwnership.LOCK_FILE); + Files.writeString(lockPath, ""); + assertEquals(SessionOwnership.State.UNOWNED, + SessionOwnership.probe(session)); + + try (FileChannel channel = FileChannel.open( + lockPath, StandardOpenOption.READ, StandardOpenOption.WRITE); + FileLock ignored = channel.lock()) { + assertEquals(SessionOwnership.State.ACTIVE, + SessionOwnership.probe(session)); + } + + assertEquals(SessionOwnership.State.UNOWNED, + SessionOwnership.probe(session)); + } + + @Test + void transportLossMarkerIsDurableCompletionGate(@TempDir Path session) + throws Exception { + assertFalse(SessionOwnership.hasTransportLoss(session)); + Files.writeString( + session.resolve(".gpufl-transport-loss.device.9.json"), "{}"); + assertTrue(SessionOwnership.hasTransportLoss(session)); + } +} diff --git a/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java b/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java index cb3d93e..6dbbab9 100644 --- a/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java +++ b/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java @@ -18,18 +18,28 @@ void booleanFlags_offByDefault_onViaFlagOrEnv() { assertFalse(ConfigLoader.parseIgnorePreexisting(new String[]{}, NO_ENV)); assertFalse(ConfigLoader.parsePruneFailed(new String[]{}, NO_ENV)); assertFalse(ConfigLoader.parseExitIfEmpty(new String[]{}, NO_ENV)); + assertFalse(ConfigLoader.parseRetainAcknowledgedPayloads( + new String[]{}, NO_ENV)); // on via flag assertTrue(ConfigLoader.parsePruneFailed(new String[]{"--prune-failed=1"}, NO_ENV)); assertTrue(ConfigLoader.parseExitIfEmpty(new String[]{"--exit-if-empty=true"}, NO_ENV)); + assertTrue(ConfigLoader.parseRetainAcknowledgedPayloads( + new String[]{"--retain-acked-payloads=1"}, NO_ENV)); // on via env assertTrue(ConfigLoader.parseExitWhenDrained(new String[]{}, Map.of("GPUFL_AGENT_EXIT_WHEN_DRAINED", "1"))); assertTrue(ConfigLoader.parseIgnorePreexisting(new String[]{}, Map.of("GPUFL_AGENT_IGNORE_PREEXISTING", "1"))); + assertTrue(ConfigLoader.parseRetainAcknowledgedPayloads( + new String[]{}, + Map.of("GPUFL_AGENT_RETAIN_ACKED_PAYLOADS", "true"))); // explicit false / 0 -> off assertFalse(ConfigLoader.parsePruneFailed(new String[]{"--prune-failed=false"}, NO_ENV)); assertFalse(ConfigLoader.parseExitIfEmpty(new String[]{}, Map.of("GPUFL_AGENT_EXIT_IF_EMPTY", "0"))); + assertFalse(ConfigLoader.parseRetainAcknowledgedPayloads( + new String[]{}, + Map.of("GPUFL_AGENT_RETAIN_ACKED_PAYLOADS", "false"))); } @Test diff --git a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java index 5fb8fb2..5de90db 100644 --- a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java +++ b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java @@ -2,6 +2,7 @@ import com.gpuflight.agent.config.HttpConfig; import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.model.WindowMetadata; import com.sun.net.httpserver.HttpServer; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -31,6 +32,7 @@ class HttpPublisherTest { private final AtomicReference lastContentType = new AtomicReference<>(); private final AtomicReference lastPath = new AtomicReference<>(); private final AtomicReference lastSessionId = new AtomicReference<>(); + private final AtomicReference responseBody = new AtomicReference<>(""); @BeforeEach void startServer() throws IOException { @@ -47,7 +49,9 @@ void startServer() throws IOException { lastContentType.set(headers.getFirst("Content-Type")); lastPath.set(exchange.getRequestURI().getPath()); lastSessionId.set(headers.getFirst("X-GpuFlight-Session-Id")); - exchange.sendResponseHeaders(statusToReturn.get(), 0); + byte[] response = responseBody.get().getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(statusToReturn.get(), response.length); + exchange.getResponseBody().write(response); exchange.close(); }); server.start(); @@ -88,6 +92,14 @@ private static byte[] gzipBytes(String s) throws IOException { return out.toByteArray(); } + private static WindowMetadata windowMetadata(byte[] body) { + return new WindowMetadata( + 1, "transport_window", + "10000000-0000-4000-8000-000000000001", + "session-1", "device", 1, 0, 1, 1, + "device.1.log.gz", body.length, 123); + } + @Test void publish_sendsPostRequest() throws InterruptedException { HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5); @@ -205,6 +217,61 @@ void publishStream_alreadyUploaded409_advancesAsAccepted() { assertTrue(ok, "409 already-uploaded must advance the cursor, not retry forever"); } + @Test + void publishWindow_identityConflict409NeverAdvances() throws IOException { + statusToReturn.set(409); + responseBody.set("{\"code\":\"window_identity_conflict\"}"); + HttpConfig config = new HttpConfig( + hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + byte[] body = gzipBytes("{\"type\":\"kernel_event\"}\n"); + WindowMetadata metadata = new WindowMetadata( + 1, "transport_window", + "10000000-0000-4000-8000-000000000001", + "session-1", "device", 1, 0, 1, 1, + "device.1.log.gz", body.length, 123); + + boolean ok = pub.publishStreamGz("session-1", metadata, body); + + assertFalse(ok, "identity conflict must preserve the payload and cursor"); + } + + @Test + void publishWindow_oldBackend202AcceptsButCannotAuthorizeDeletion() + throws IOException { + statusToReturn.set(202); + responseBody.set( + "{\"accepted_for_processing\":true,\"spool_id\":\"legacy\"}"); + HttpPublisher pub = new HttpPublisher(new HttpConfig( + hostUrl(), "v1", null, 5, + "stream", 100, 1_000_000L)); + byte[] body = gzipBytes("{\"type\":\"kernel_event\"}\n"); + + WindowPublishResult result = pub.publishTransportWindow( + "session-1", windowMetadata(body), body); + + assertTrue(result.accepted()); + assertFalse(result.identityAcknowledged(), + "legacy 2xx must retain the local replayable payload"); + } + + @Test + void publishWindow_newBackendAckAuthorizesDeletion() + throws IOException { + statusToReturn.set(202); + responseBody.set("{\"window_acknowledged\":true}"); + HttpPublisher pub = new HttpPublisher(new HttpConfig( + hostUrl(), "v1", null, 5, + "stream", 100, 1_000_000L)); + byte[] body = gzipBytes("{\"type\":\"kernel_event\"}\n"); + + WindowPublishResult result = pub.publishTransportWindow( + "session-1", windowMetadata(body), body); + + assertTrue(result.accepted()); + assertTrue(result.identityAcknowledged()); + } + @Test void publishStream_limitExceeded402_advancesAsAccepted() { // 402 (workspace/GPU limit) is permanent; retrying won't help. diff --git a/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java b/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java index d679a88..8b8e9c2 100644 --- a/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java +++ b/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java @@ -69,6 +69,20 @@ void discoversFlatAndNestedTogether(@TempDir Path root) throws IOException { found.stream().map(DiscoveredSession::sessionId).sorted().toList()); } + @Test + void discoversSessionAfterAckCleanupLeavesOnlyWindowTombstone( + @TempDir Path root) throws IOException { + Path session = Files.createDirectories(root.resolve("acked-not-settled")); + Files.writeString( + session.resolve(".gpufl-window.device.1.json"), "{}"); + + List found = + SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertEquals(1, found.size()); + assertEquals("acked-not-settled", found.get(0).sessionId()); + } + @Test void ignoresDirsWithoutChannelFilesAndDotDirs(@TempDir Path root) throws IOException { Files.createDirectories(root.resolve("not-a-session").resolve("random")); diff --git a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java index ff4407d..4393dfd 100644 --- a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java +++ b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java @@ -34,12 +34,37 @@ static final class NoopPublisher implements Publisher { @Override public void close() {} } + static final class CompletionPublisher implements Publisher { + volatile int completionSignals; + @Override public boolean publish(String topic, String key, LogWrapper log) { + return true; + } + @Override public boolean publishStream( + String sessionId, List ndjsonLines) { + return true; + } + @Override public boolean publishSessionComplete(String sessionId) { + completionSignals++; + return true; + } + @Override public void close() {} + } + private TailerManager managerIn(ExecutorService ex, Path root) { return new TailerManager(ex, new NoopPublisher(), new CursorManager(new File(root.toFile(), "cursor.json")), new LinkedBlockingQueue<>(), null, StreamUploadSettings.DISABLED, "gpu-trace"); } + private TailerManager streamManagerIn( + ExecutorService ex, Path root, Publisher publisher) { + return new TailerManager(ex, publisher, + new CursorManager(new File(root.toFile(), "cursor.json")), + new LinkedBlockingQueue<>(), null, + new StreamUploadSettings(true, 100, 1_000_000L), + "gpu-trace"); + } + private static void runToCompletion(ExecutorService ex) throws InterruptedException { ex.shutdown(); assertTrue(ex.awaitTermination(20, TimeUnit.SECONDS), "tailers did not finish"); @@ -91,4 +116,27 @@ void orphanedFinishPrunesWhenEnabled(@TempDir Path root) throws Exception { assertFalse(sess.exists(), "orphaned session should be pruned"); } + + @Test + void transportLossNeverSignalsSessionComplete(@TempDir Path root) + throws Exception { + File sess = new File(root.toFile(), "sess-loss"); + Files.createDirectories(sess.toPath()); + Files.writeString(new File(sess, "device.1.log").toPath(), + "{\"type\":\"shutdown\",\"session_id\":\"sess-loss\"}\n"); + Files.writeString( + new File(sess, + ".gpufl-transport-loss.device.1.json").toPath(), "{}"); + + CompletionPublisher publisher = new CompletionPublisher(); + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + streamManagerIn(ex, root, publisher).spawnSessionTailers( + new DiscoveredSession( + root.toFile(), "sess-loss", List.of("device"))); + runToCompletion(ex); + + assertEquals(0, publisher.completionSignals); + assertTrue(new File(sess, SessionWatcher.FAILED_MARKER).exists()); + assertFalse(new File(sess, SessionWatcher.UPLOADED_MARKER).exists()); + } }