JSONLogger: rate-limit resProgress per activity

src/libutil-tests/json-logger.cc

@@ -0,0 +1,47 @@
+#include "nix/util/logging.hh"
+#include "nix/util/file-system.hh"
+#include "nix/util/strings.hh"
+
+#include <gtest/gtest.h>
+
+namespace nix {
+
+static size_t countLines(const std::filesystem::path & path, std::string_view needle)
+{
+    size_t n = 0;
+    for (auto & line : tokenizeString<Strings>(readFile(path), "\n"))
+        if (line.find(needle) != line.npos)
+            n++;
+    return n;
+}
+
+TEST(JSONLogger, progressIsRateLimited)
+{
+    auto tmpDir = createTempDir();
+    AutoDelete delTmpDir(tmpDir, true);
+    auto path = tmpDir / "log";
+    {
+        auto jsonLogger = makeJSONLogger(path, false);
+        Activity act(*jsonLogger, lvlInfo, actCopyPath, "copy", {}, 0);
+        for (uint64_t i = 1; i <= 100000; ++i)
+            act.progress(i, 100000);
+    }
+    EXPECT_LT(countLines(path, "\"type\":105"), 100u);
+    EXPECT_EQ(countLines(path, "\"fields\":[100000,100000,0,0]"), 1u);
+}
+
+TEST(JSONLogger, otherResultsAreNotRateLimited)
+{
+    auto tmpDir = createTempDir();
+    AutoDelete delTmpDir(tmpDir, true);
+    auto path = tmpDir / "log";
+    {
+        auto jsonLogger = makeJSONLogger(path, false);
+        Activity act(*jsonLogger, lvlInfo, actBuild, "build", {}, 0);
+        for (int i = 0; i < 1000; ++i)
+            act.result(resBuildLogLine, "x");
+    }
+    EXPECT_EQ(countLines(path, "\"type\":101"), 1000u);
+}
+
+} // namespace nix

src/libutil-tests/meson.build

@@ -59,6 +59,7 @@ sources = files(
   'git.cc',
   'hash.cc',
   'hilite.cc',
+  'json-logger.cc',
   'json-utils.cc',
   'local-keys.cc',
   'logging.cc',

src/libutil/logging.cc

@@ -9,6 +9,8 @@
 #include "nix/util/unix-domain-socket.hh"
 
 #include <atomic>
+#include <chrono>
+#include <map>
 #include <sstream>
 #include <nlohmann/json.hpp>
 
@@ -253,10 +255,28 @@ struct JSONLogger : Logger
     struct State
     {
         bool enabled = true;
+
+        /* resProgress is sampled per activity so that per-chunk updates
+           (e.g. from copyPaths) cannot back-pressure the log fd. */
+        std::map<ActivityId, std::chrono::steady_clock::time_point> lastProgress;
     };
 
     Sync<State> _state;
 
+    static constexpr std::chrono::milliseconds progressInterval{100};
+
+    /* resProgress fields are [done, expected, running, failed]. The update
+       that reaches `expected` is never dropped by the rate limit, otherwise
+       consumers could be left showing e.g. 97% for a finished copy. */
+    static bool isFinalProgress(std::span<const Field> fields)
+    {
+        if (fields.size() < 2)
+            return false;
+        auto done = std::get_if<uint64_t>(&fields[0]);
+        auto expected = std::get_if<uint64_t>(&fields[1]);
+        return done && expected && *expected && *done >= *expected;
+    }
+
     void write(const nlohmann::json & json)
     {
         auto line = (includeNixPrefix ? "@nix " : "")
@@ -335,6 +355,8 @@ struct JSONLogger : Logger
 
     void stopActivity(ActivityId act) noexcept override
     {
+        _state.lock()->lastProgress.erase(act);
+
         nlohmann::json json;
         json["action"] = "stop";
         json["id"] = act;
@@ -348,6 +370,15 @@ struct JSONLogger : Logger
         json["id"] = act;
         json["type"] = type;
         addFields(json, fields);
+
+        if (type == resProgress && !isFinalProgress(fields)) {
+            auto now = std::chrono::steady_clock::now();
+            auto & last = _state.lock()->lastProgress[act];
+            if (now - last < progressInterval)
+                return;
+            last = now;
+        }
+
         write(json);
     }
 };