Add crash-safe transport-window rotation

CMakeLists.txt

@@ -155,6 +155,8 @@ target_sources(gpufl PRIVATE
     include/gpufl/core/logger/logger.cpp
     include/gpufl/core/logger/log_rotator.cpp
     include/gpufl/core/logger/log_salvage.cpp
+    include/gpufl/core/logger/session_ownership.cpp
+    include/gpufl/core/logger/window_metadata.cpp
     include/gpufl/core/logger/file_log_sink.cpp
     include/gpufl/upload/upload_logs.cpp
     include/gpufl/core/host_info.cpp

daemon/launcher/trace_command_common.cpp

@@ -288,37 +288,21 @@ bool appendLineToGzipLog(const fs::path& gz_path, const std::string& line) {
     return true;
 }
 
-// Highest existing system.<k>.log[.gz] window index under session_dir (0 if none).
-int highestSystemWindowIndex(const fs::path& session_dir) {
-    int max_idx = 0;
-    std::error_code ec;
-    for (const auto& entry : fs::directory_iterator(session_dir, ec)) {
-        if (ec) break;
-        if (!entry.is_regular_file(ec)) continue;
-        std::string name = entry.path().filename().string();
-        if (name.size() > 3 && name.compare(name.size() - 3, 3, ".gz") == 0) {
-            name.resize(name.size() - 3);
-        }
-        if (name.size() <= 4 || name.compare(name.size() - 4, 4, ".log") != 0) continue;
-        name.resize(name.size() - 4);                  // "system" or "system.<k>"
-        if (name.rfind("system.", 0) != 0) continue;   // skip the non-indexed "system"
-        const std::string suffix = name.substr(std::string("system.").size());
-        if (suffix.empty() ||
-            !std::all_of(suffix.begin(), suffix.end(),
-                         [](unsigned char c) { return std::isdigit(c); })) {
-            continue;
-        }
-        try { max_idx = std::max(max_idx, std::stoi(suffix)); } catch (...) {}
-    }
-    return max_idx;
-}
-
 bool appendSyntheticShutdown(const fs::path& session_dir,
                              const std::string& session_id,
                              const SessionLifecycleInfo& info,
                              const SyntheticShutdownContext& context) {
     const std::string line = syntheticShutdownLine(session_id, info, context);
-    const std::string base = "system." + std::to_string(highestSystemWindowIndex(session_dir) + 1) + ".log";
+    // Allocate through the shared allocator, which also counts windows still
+    // sitting in `.tmp`. highestSystemWindowIndex() only scans the session
+    // ROOT, so a deferred window in `.tmp` did not reserve its index and the
+    // marker could be written onto it - one of the two then gets deleted as
+    // a "duplicate", losing either a full window or the shutdown record the
+    // backend needs to finalize the session.
+    const std::string base =
+        "system." +
+        std::to_string(gpufl::nextLogWindowIndex(session_dir, "system")) +
+        ".log";
     const fs::path window_gz = session_dir / (base + ".gz");
     std::error_code ec;
     if (fs::exists(window_gz, ec)) {
@@ -425,6 +409,17 @@ void signalSessionsComplete(const fs::path& output_dir,
         if (!isSessionDirectory(entry)) continue;
 
         const std::string session_id = entry.path().filename().string();
+        const auto lost_windows = transportLossMarkerCount(entry.path());
+        if (lost_windows > 0) {
+            if (!quiet) {
+                std::fprintf(
+                    stderr,
+                    "gpufl trace --upload: NOT signalling session-complete "
+                    "for %s: %zu transport window(s) are known lost\n",
+                    session_id.c_str(), lost_windows);
+            }
+            continue;
+        }
         futures.push_back(std::async(std::launch::async, [config, session_id]() {
             return std::make_pair(session_id, postSessionComplete(config, session_id));
         }));
@@ -486,6 +481,19 @@ int repairUncompressedLogs(const fs::path& root) {
 
         const fs::path gz_path(path.string() + ".gz");
         std::error_code exists_ec;
+        // EXISTS is not VALID. A compress that died midway - or a rename that
+        // hit the disk before its data - leaves a truncated or empty `.gz`,
+        // and deleting the raw because that file is merely present destroys
+        // the only complete copy of the window. Re-compress from the raw
+        // instead when the `.gz` does not decode.
+        if (fs::exists(gz_path, exists_ec) && !isValidGzipFile(gz_path)) {
+            std::fprintf(stderr,
+                         "[gpufl] warning: %s is not a readable gzip - "
+                         "rebuilding it from %s\n",
+                         gz_path.string().c_str(), path.string().c_str());
+            std::error_code rm_ec;
+            fs::remove(gz_path, rm_ec);
+        }
         if (fs::exists(gz_path, exists_ec)) {
             std::error_code remove_ec;
             if (!removeWithRetry(path, remove_ec)) {

include/gpufl/core/env_vars.hpp

@@ -82,6 +82,20 @@ constexpr const char* kDebugOutput          = "GPUFL_DEBUG";
 // exercises rotation without writing tens of MB.
 constexpr const char* kLogRotateBytes       = "GPUFL_LOG_ROTATE_BYTES";
 
+// Opt-in (>0): also rotate a channel window once the data in it spans this
+// many milliseconds (monotonic clock, measured from the window's first
+// write). Publishes transport windows on a time cadence so low-volume
+// channels stop sitting in `.tmp` until 64 MiB accumulates - the agent
+// ships each finished window mid-run. An empty window is never rotated.
+constexpr const char* kLogRotateAfterMs     = "GPUFL_LOG_ROTATE_AFTER_MS";
+
+// Hard safety limits for the per-session transport spool. Once either limit
+// is reached, GPUFlight stops accepting new profiling events and writes a
+// durable transport-loss marker instead of filling the application's disk.
+// The marker prevents a later uploader from reporting the session complete.
+constexpr const char* kLogMaxSpoolBytes     = "GPUFL_LOG_MAX_SPOOL_BYTES";
+constexpr const char* kLogMinFreeBytes      = "GPUFL_LOG_MIN_FREE_BYTES";
+
 // Opt-in ("1", "true", "yes", "on"): flush each log line immediately.
 // Useful when diagnosing whether missing records are buffered in userspace.
 constexpr const char* kFlushLogsAlways      = "GPUFL_FLUSH_LOGS_ALWAYS";

include/gpufl/core/gpufl.cpp

@@ -331,6 +331,19 @@ bool init(const InitOptions& opts) {
             logOpts.rotate_bytes = static_cast<std::size_t>(bytes);
         }
     }
+    if (const char* v = std::getenv(env::kLogRotateAfterMs)) {
+        if (const auto ms = std::strtoll(v, nullptr, 10); ms > 0) {
+            logOpts.rotate_after_ms = static_cast<std::int64_t>(ms);
+        }
+    }
+    if (const char* v = std::getenv(env::kLogMaxSpoolBytes)) {
+        logOpts.max_spool_bytes =
+            static_cast<std::uint64_t>(std::strtoull(v, nullptr, 10));
+    }
+    if (const char* v = std::getenv(env::kLogMinFreeBytes)) {
+        logOpts.min_free_bytes =
+            static_cast<std::uint64_t>(std::strtoull(v, nullptr, 10));
+    }
 
     g_lastLogPath = logPath;
     g_lastSessionId = rt->session_id;

include/gpufl/core/logger/file_compressor.cpp

@@ -28,6 +28,24 @@ bool removeWithRetry(const fs::path& p, std::error_code& ec) {
 }
 }  // namespace
 
+bool removeOrTruncateFile(const std::string& path) {
+    if (path.empty()) return true;
+
+    const fs::path p(path);
+    std::error_code ec;
+    if (removeWithRetry(p, ec)) return true;
+
+    // Windows can deny delete sharing while still allowing the owner to
+    // truncate. An empty husk is harmless and the temp-dir cleanup removes it
+    // once the holder lets go.
+    std::ofstream trunc(p, std::ios::out | std::ios::trunc);
+    if (!trunc) return false;
+    trunc.close();
+
+    std::error_code size_ec;
+    return fs::file_size(p, size_ec) == 0 && !size_ec;
+}
+
 bool GzipFileCompressor::compressTo(const std::string& src,
                                     const std::string& dst) {
     if (src.empty() || dst.empty()) return false;
@@ -71,25 +89,17 @@ bool GzipFileCompressor::compress(const std::string& path) {
     const bool ok = compressTo(path, outPath);
     if (!ok) return false;
 
-    std::error_code ec;
-    if (!removeWithRetry(path, ec)) {
-        // A holder outlived the retries (a tail, an editor, an AV scan).
-        // Windows lets nobody delete a file held without delete sharing -
-        // but write sharing is common, so TRUNCATE the original instead:
-        // the data then exists exactly once (in the .gz) and the leftover
-        // is an empty husk that any later cleanup removes once the holder
-        // lets go.
-        std::ofstream trunc(path, std::ios::out | std::ios::trunc);
-        if (trunc) {
-            GFL_LOG_DEBUG("[Logger] compressed '", path,
-                          "' - original was held, truncated to empty "
-                          "instead of removed.");
-        } else {
-            GFL_LOG_ERROR("[Logger] compressed '", path,
-                          "' but could not remove or truncate the original "
-                          "(", ec.message(),
-                          ") - stale .log left next to the .gz.");
-        }
+    if (!removeOrTruncateFile(path)) {
+        // Keep the source authoritative when the exact-once transition could
+        // not complete. A caller seeing false may retry safely; leaving both
+        // a non-empty raw file and a successful gzip would invite duplicate
+        // salvage/upload.
+        std::error_code rm_ec;
+        fs::remove(outPath, rm_ec);
+        GFL_LOG_ERROR("[Logger] compressed '", path,
+                      "' but could not remove or truncate the original; "
+                      "discarded the gzip and kept the raw source.");
+        return false;
     }
     return true;
 }

include/gpufl/core/logger/file_compressor.hpp

@@ -4,6 +4,16 @@
 
 namespace gpufl {
 
+/**
+ * Remove a file with the short Windows sharing-violation retry used by the
+ * logger, falling back to truncation when a holder prevents unlinking.
+ *
+ * Returns true only when the path is gone or is an empty husk. Callers use
+ * this to preserve the single-authority spool contract: a completed gzip
+ * must never be published while an identical non-empty raw window remains.
+ */
+bool removeOrTruncateFile(const std::string& path);
+
 class IFileCompressor {
    public:
     virtual ~IFileCompressor() = default;