From f15165e7b31ab1380c732268baf63cba11c64d38 Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Wed, 17 Jun 2026 22:45:15 -0700 Subject: [PATCH] retry handling for already existing error. --- .../java/com/gpuflight/agent/LogTailer.java | 16 ++-- src/main/java/com/gpuflight/agent/Main.java | 43 +++++++++++ .../agent/publisher/HttpPublisher.java | 56 ++++++++++++++ .../gpuflight/agent/publisher/Publisher.java | 15 ++++ .../com/gpuflight/agent/LogTailerTest.java | 49 ++++++++++++ .../agent/publisher/HttpPublisherTest.java | 75 +++++++++++++++++++ 6 files changed, 248 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/gpuflight/agent/LogTailer.java b/src/main/java/com/gpuflight/agent/LogTailer.java index 3dd4386..bea17a4 100644 --- a/src/main/java/com/gpuflight/agent/LogTailer.java +++ b/src/main/java/com/gpuflight/agent/LogTailer.java @@ -385,9 +385,12 @@ public void tail(Publisher publisher) { continue; } if (!activeFile().exists()) { - idx = 0; - offset = 0; - cursorMgr.update(streamKey, idx, offset); + // Session finished. Leave the cursor at this drained file's + // EOF (drainGz already persisted {idx, endOffset, identity}) + // so a restart resumes HERE, skips to EOF and exits. Do NOT + // reset to offset 0 - that keeps the file identity but wipes + // progress, so the next run re-locates this file by content + // signature and re-uploads the whole session (backend 409s it). System.out.println("[" + logType + "] No active file remains; session finished."); return; } @@ -514,9 +517,10 @@ else if (rotated != null && (isGz(rotated) || rotated.length() >= offset)) { continue; } if (!activeFile().exists()) { - idx = 0; - offset = 0; - cursorMgr.update(streamKey, idx, offset); + // Session finished. Leave the cursor at this drained + // file's EOF (updated above to {idx, offset, identity}) + // so a restart skips to EOF and exits. Do NOT reset to + // offset 0 - that re-uploads the whole session next run. System.out.println("[" + logType + "] No active file remains; session finished."); return; } diff --git a/src/main/java/com/gpuflight/agent/Main.java b/src/main/java/com/gpuflight/agent/Main.java index 45f316d..9c7b9ec 100644 --- a/src/main/java/com/gpuflight/agent/Main.java +++ b/src/main/java/com/gpuflight/agent/Main.java @@ -102,6 +102,15 @@ static void main(String[] args) throws Exception { if (!startedSessions.add(key)) return; System.out.println("[agent] Tailing session \"" + session.sessionId() + "\" in " + session.folder() + " types=" + session.logTypes()); + // Count this session's per-channel tailers. When the LAST one finishes + // NATURALLY (session drained + every batch accepted), tell the backend the + // upload is complete so it can finalize without waiting out its grace + // window. A tailer that exits via interruption (agent shutdown) or never + // finishes (a channel whose file never appears) does NOT count toward the + // signal — so a partial upload is never declared complete; the backend's + // grace path covers those. + var remaining = new java.util.concurrent.atomic.AtomicInteger(session.logTypes().size()); + String sid = session.sessionId(); for (String type : session.logTypes()) { executor.submit(() -> { // Only the "system" channel carries device_metric_batch events. @@ -110,6 +119,13 @@ static void main(String[] args) throws Exception { topicPrefix, cursorMgr, consumedFilesQueue, dedup, streamUploadSettings); tailer.tail(publisher); + // Reached only when tail() returns. Skip on interruption (agent + // shutdown) so an in-flight upload is never declared complete. + if (!Thread.currentThread().isInterrupted() + && remaining.decrementAndGet() == 0 + && streamUploadSettings.enabled()) { + signalSessionComplete(publisher, sid); + } }); } }; @@ -176,6 +192,33 @@ static void main(String[] args) throws Exception { new CountDownLatch(1).await(); } + /** + * Tell the backend that every channel of a session has finished uploading. + * Best-effort with a few bounded retries: {@code publishSessionComplete} + * returns true for terminal outcomes (2xx, or a 4xx like 404 from an older + * backend) and false only for transient 5xx/network errors. If it never gets + * through, the backend's grace path still finalizes the session, so this is + * a pure latency optimization — never a correctness dependency. + */ + private static void signalSessionComplete(Publisher publisher, String sessionId) { + for (int attempt = 1; attempt <= 3; attempt++) { + try { + if (publisher.publishSessionComplete(sessionId)) return; + } catch (Exception e) { + System.err.println("[agent] session-complete signal error for " + + sessionId + ": " + e.getMessage()); + } + try { + Thread.sleep(2000L); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } + } + System.out.println("[agent] session-complete signal gave up for " + sessionId + + " - backend grace finalize will apply"); + } + // ------------------------------------------------------------------------- // Config resolution // ------------------------------------------------------------------------- diff --git a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java index d2979ae..00f1cf2 100644 --- a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java +++ b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java @@ -106,6 +106,23 @@ public boolean publishStream(String sessionId, List ndjsonLines) { + response.statusCode() + " session=" + sessionId); return true; } + if (response.statusCode() == 409) { + // Session already finalized on the backend - typically this agent + // re-tailing a finished session after a restart. The data is + // already stored and re-profiling is required to replace it, so + // retrying NEVER succeeds. Advance like a 2xx (mirrors the 402 + // handling in publish()) so the tailer drains to EOF and exits + // instead of re-POSTing the same batch every 5s forever. + System.out.println("[agent] HTTP stream POST: session already uploaded (409) - " + + "skipping, session=" + sessionId); + return true; + } + if (response.statusCode() == 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; + } System.out.println("HTTP stream publish failed [" + url + "]: " + response.statusCode() + " - " + response.body()); return false; @@ -118,6 +135,45 @@ public boolean publishStream(String sessionId, List ndjsonLines) { } } + @Override + public boolean publishSessionComplete(String sessionId) { + try { + String url = config.endpointFor("session-complete"); + System.out.println("[agent] HTTP session-complete POST starting: url=" + url + + " session=" + sessionId); + + var requestBuilder = HttpRequest.newBuilder() + .uri(URI.create(url)) + .header("X-GpuFlight-Session-Id", sessionId) + .POST(HttpRequest.BodyPublishers.noBody()); + + addAuthHeader(requestBuilder); + + var response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString()); + int sc = response.statusCode(); + if (sc >= 200 && sc <= 299) { + System.out.println("[agent] HTTP session-complete accepted: status=" + sc + + " session=" + sessionId); + return true; + } + if (sc >= 400 && sc < 500) { + // Old backend without the endpoint (404), or a client error: nothing + // to retry — the backend's grace path still finalizes the session. + System.out.println("[agent] HTTP session-complete not applicable (status=" + sc + + ") session=" + sessionId + " - backend grace finalize will apply"); + return true; + } + System.out.println("HTTP session-complete failed [" + url + "]: " + sc + " - " + response.body()); + return false; // 5xx: transient, worth a bounded retry + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } catch (Exception e) { + System.out.println("HTTP session-complete connection error: " + e); + return false; + } + } + private void addAuthHeader(HttpRequest.Builder requestBuilder) { if (config.authToken() != null && !config.authToken().isBlank()) { requestBuilder.header("Authorization", "Bearer " + config.authToken()); diff --git a/src/main/java/com/gpuflight/agent/publisher/Publisher.java b/src/main/java/com/gpuflight/agent/publisher/Publisher.java index 8228d41..ea530ac 100644 --- a/src/main/java/com/gpuflight/agent/publisher/Publisher.java +++ b/src/main/java/com/gpuflight/agent/publisher/Publisher.java @@ -13,4 +13,19 @@ public interface Publisher extends Closeable { default boolean publishStream(String sessionId, List ndjsonLines) { return false; } + + /** + * Signal the backend that EVERY channel of {@code sessionId} has finished + * uploading — the agent has drained all per-channel tailers and every batch + * was accepted. Lets the backend finalize the session immediately instead of + * waiting out its conservative grace/settle window. + * + *

Returns true when the signal was delivered OR is pointless to retry + * (2xx, or a 4xx like 404 from an older backend without the endpoint); false + * only on a transient failure (5xx / network) worth retrying. Default no-op + * (e.g. Kafka, where there is no such finalize step) returns true. + */ + default boolean publishSessionComplete(String sessionId) { + return true; + } } diff --git a/src/test/java/com/gpuflight/agent/LogTailerTest.java b/src/test/java/com/gpuflight/agent/LogTailerTest.java index bdd2a72..94af7a0 100644 --- a/src/test/java/com/gpuflight/agent/LogTailerTest.java +++ b/src/test/java/com/gpuflight/agent/LogTailerTest.java @@ -632,4 +632,53 @@ void tail_compressedActive_resumesFromOffset(@TempDir Path tempDir) throws Excep assertEquals(1, publisher.events.size(), "Only the tail after the offset should be read"); assertTrue(publisher.events.get(0).data().contains("k2")); } + + // ---- A finished session is not re-uploaded when the agent restarts ---- + + @Test + void tail_finishedRotatedGzSession_notReuploadedOnRestart(@TempDir Path tempDir) throws Exception { + // A finished session whose channel rotated: only .1.log.gz + // remains (no active .log). Run 1 drains it fully and tail() + // returns. Run 2 over the SAME files + cursor must NOT re-publish: the + // cursor left at the gz's EOF makes the restart skip-to-EOF and exit. + // + // Regression: the terminal "session finished" branch used to reset the + // cursor to offset 0 while keeping the file identity, so run 2 re-located + // the gz by content signature and re-uploaded the whole session - which + // the backend then rejected with 409 "session already uploaded", looping + // forever. + Files.createDirectories(tempDir.resolve("app")); + String content = + "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" + + "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n"; + Path plain = tempDir.resolve("app/device.1.log"); + Files.writeString(plain, content); + Path gz = tempDir.resolve("app/device.1.log.gz"); + gzip(plain, gz); + Files.delete(plain); // finished: only the rotated .gz remains, no active .log + + File cursorFile = tempDir.resolve("cursor.json").toFile(); + + // Run 1: drains the gz fully; tail() returns on "session finished". + CapturingStreamPublisher pub1 = new CapturingStreamPublisher(); + LogTailer tailer1 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", + new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null, + new StreamUploadSettings(true, 10, 1_000_000L)); + Thread t1 = startTailer(tailer1, pub1); + awaitEvents(pub1.batches, 1, 3000); + t1.join(3000); // tail() exits by itself once the finished session is drained + assertFalse(t1.isAlive(), "tailer should exit on its own after draining a finished session"); + assertEquals(1, pub1.batches.size(), "run 1 publishes the finished session once"); + + // Run 2: same files, same persisted cursor. Must NOT re-publish anything. + CapturingStreamPublisher pub2 = new CapturingStreamPublisher(); + LogTailer tailer2 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", + new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null, + new StreamUploadSettings(true, 10, 1_000_000L)); + Thread t2 = startTailer(tailer2, pub2); + t2.join(3000); // should skip-to-EOF and exit immediately + assertFalse(t2.isAlive(), "restart of a finished session should exit immediately"); + assertTrue(pub2.batches.isEmpty(), + "a finished session must NOT be re-uploaded on restart (got " + pub2.batches + ")"); + } } diff --git a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java index ff245a5..2fa1536 100644 --- a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java +++ b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java @@ -143,6 +143,48 @@ void publishStream_sendsGzippedNdjsonBatch() { assertTrue(lastBody.get().endsWith("\n")); } + @Test + void publishStream_alreadyUploaded409_advancesAsAccepted() { + // 409 "session already uploaded" is terminal - retrying never succeeds. + // publishStream must return true (advance) so the tailer drains to EOF and + // exits instead of re-POSTing the same batch every 5s forever. + statusToReturn.set(409); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + + boolean ok = pub.publishStream("session-1", + List.of("{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}")); + + assertTrue(ok, "409 already-uploaded must advance the cursor, not retry forever"); + } + + @Test + void publishStream_limitExceeded402_advancesAsAccepted() { + // 402 (workspace/GPU limit) is permanent; retrying won't help. + statusToReturn.set(402); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + + boolean ok = pub.publishStream("session-1", + List.of("{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}")); + + assertTrue(ok, "402 limit-exceeded must advance, not retry forever"); + } + + @Test + void publishStream_serverError500_signalsRetry() { + // A transient 5xx must remain retriable (return false) so the tailer backs + // off and retries from the same offset. + statusToReturn.set(500); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + + boolean ok = pub.publishStream("session-1", + List.of("{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}")); + + assertFalse(ok, "5xx is transient and must be retried"); + } + @Test void publish_nonSuccessResponse_doesNotThrow() throws InterruptedException { statusToReturn.set(500); @@ -169,4 +211,37 @@ void close_doesNotThrow() { HttpPublisher pub = new HttpPublisher(config); assertDoesNotThrow(pub::close); } + + // ── publishSessionComplete (agent "all channels uploaded" signal) ── + + @Test + void publishSessionComplete_postsToEndpointWithSessionHeader_andReturnsTrueOn200() { + statusToReturn.set(200); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5); + HttpPublisher pub = new HttpPublisher(config); + + boolean ok = pub.publishSessionComplete("session-xyz"); + + assertTrue(ok); + assertEquals("/api/v1/events/session-complete", lastPath.get()); + assertEquals("session-xyz", lastSessionId.get()); + } + + @Test + void publishSessionComplete_oldBackend404_returnsTrueSoAgentDoesNotRetry() { + // An older backend without the endpoint 404s; the agent must treat it as + // terminal (the grace path still finalizes the session) and not loop. + statusToReturn.set(404); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5); + boolean ok = new HttpPublisher(config).publishSessionComplete("session-xyz"); + assertTrue(ok); + } + + @Test + void publishSessionComplete_serverError500_returnsFalseForRetry() { + statusToReturn.set(500); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5); + boolean ok = new HttpPublisher(config).publishSessionComplete("session-xyz"); + assertFalse(ok); + } }