nix-eval-jobs: reenable tests, rework waitpid handling

Closes #703

The problem boils down to:

* A testcase with a segfaulting worker due to a stack overflow from the
  evaluation.

* When the controller encounters that the pipes are down, the worker's
  state is checked and a potential infrec is reported if it stopped with
  SIGSEGV.

* However, in a few cases, the worker didn't exit yet, but only had closed
  pipes (which is how the error is detected). In that case, a different
  error message is printed breaking the test.

This bug handling only affects error cases, so this race doesn't happen
on normal shutdown. To mitigate the problem a bit, the controller will
wait for a second and re-check the state of the worker process then.

While this should make the error reporting for end-users a little more
reliable, it's still no guarantee for a non-racy test. Hence, the test
asserts against both cases now and it passes if either the infrec error
or the "worker still running, but pipes are closed" error is returned.

Change-Id: Ifdc7a05bc86b6aecd5d03118d3e3ffc9affe1c5e

flake.nix

@@ -293,8 +293,7 @@
 
         # System tests.
         tests = import ./tests/nixos { inherit lib nixpkgs nixpkgsFor; } // {
-          # this test is *incredibly* flaky, sometimes taking six tries in ci to unflake
-          # nix-eval-jobs = forAllSystems (system: self.packages.${system}.nix-eval-jobs.tests.nix-eval-jobs);
+          nix-eval-jobs = forAllSystems (system: self.packages.${system}.nix-eval-jobs.tests.nix-eval-jobs);
 
           # This is x86_64-linux only, just because we have significantly
           # cheaper x86_64-linux compute in CI.

subprojects/nix-eval-jobs/src/nix-eval-jobs.cc

@@ -1,3 +1,4 @@
+#include <chrono>
 #include <lix/config.h> // IWYU pragma: keep
 
 #include <lix/libexpr/eval-settings.hh>
@@ -35,6 +36,7 @@
 #include <set>
 #include <string>
 #include <string_view>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -157,7 +159,7 @@ struct State {
     std::map<std::string, nlohmann::json> jobs;
 };
 
-void handleBrokenWorkerPipe(Proc &proc, std::string_view msg) {
+void handleBrokenWorkerPipe(Proc &proc, std::string_view msg, bool retry = true) {
     // we already took the process status from Proc, no
     // need to wait for it again to avoid error messages
     pid_t pid = proc.pid.release();
@@ -165,10 +167,21 @@ void handleBrokenWorkerPipe(Proc &proc, std::string_view msg) {
         int status;
         int rc = waitpid(pid, &status, WNOHANG);
         if (rc == 0) {
-            kill(pid, SIGKILL);
-            throw Error("BUG: while %s, worker pipe got closed but evaluation "
-                        "worker still running?",
-                        msg);
+            // If the worker dies (e.g. with a SIGSEGV due to an unnoticed infinite
+            // recursion), it closes the pipes first and then exits. Now it may happen
+            // that a read from the pipe happens when the process is still alive, but the
+            // pipes are closed.
+            // This is still a valid condition and shouldn't be reported as `BUG:`. Hence
+            // we wait a bit and then retry.
+            if (retry) {
+                std::this_thread::sleep_for(std::chrono::seconds(1));
+                handleBrokenWorkerPipe(proc, msg, false);
+            } else {
+                kill(pid, SIGKILL);
+                throw Error("BUG: while %s, worker pipe got closed but evaluation "
+                            "worker still running?",
+                            msg);
+            }
         } else if (rc == -1) {
             kill(pid, SIGKILL);
             throw Error(

subprojects/nix-eval-jobs/tests/test_eval.py

@@ -68,17 +68,17 @@ def common_test(extra_args: List[str]) -> List[Dict[str, Any]]:
         assert built_job["outputs"]["out"].startswith("/nix/store")
         assert built_job["drvPath"].endswith(".drv")
         assert built_job["meta"]["broken"] is False
-        check_gc_root(tempdir, built_job['drvPath'])
+        check_gc_root(tempdir, built_job["drvPath"])
 
         dotted_job = results[1]
         assert dotted_job["attr"] == '"dotted.attr"'
         assert dotted_job["attrPath"] == ["dotted.attr"]
-        check_gc_root(tempdir, dotted_job['drvPath'])
+        check_gc_root(tempdir, dotted_job["drvPath"])
 
         recurse_drv = results[2]
         assert recurse_drv["attr"] == "recurse.drvB"
         assert recurse_drv["name"] == "drvB"
-        check_gc_root(tempdir, recurse_drv['drvPath'])
+        check_gc_root(tempdir, recurse_drv["drvPath"])
 
         substituted_job = results[3]
         assert substituted_job["attr"] == "substitutedJob"
@@ -138,8 +138,23 @@ def test_recursion_error() -> None:
             ],
         )
         print(stderr)
-        assert "packageWithInfiniteRecursion" in stderr
-        assert "possible infinite recursion" in stderr
+
+        # Closing pipes and exiting is not atomic, so it is possible
+        # that the worker is still up when the collector notices that
+        # the pipe is closed already. We mitigate this condition a bit
+        # by waiting a second and checking the state of the worker again,
+        # but assuming the infrec error is still racy.
+        # Hence, we assert that one of the two possible outcomes actually happen.
+        assert (
+            (
+                "packageWithInfiniteRecursion" in stderr
+                and "possible infinite recursion" in stderr
+            )
+            or
+            (
+                "worker pipe got closed" in stderr
+            )
+        )
 
 
 def test_constituents() -> None: