Bound the Lean build-lock wait by the request timeout; deflake the restage test

docker/lean/server.py

@@ -120,10 +120,21 @@ def build_package(project, mods, timeout, run=subprocess.run):
     staging cannot rewrite sources under a compile. Requests for other
     projects take other locks and build in parallel. The lock is released
     before the caller elaborates: that only reads the finished oleans.
+
+    Waiting for the lock counts against `timeout`: a request that cannot get
+    it in time raises `subprocess.TimeoutExpired`, and the build gets only
+    what is left.
     """
-    with build.project_lock(os.path.join(project, "lean")):
+    lock = build.project_lock(os.path.join(project, "lean"))
+    started = time.monotonic()
+    if not lock.acquire(timeout=max(timeout, 0)):
+        raise subprocess.TimeoutExpired("build " + project, timeout)
+    try:
+        remaining = max(timeout - (time.monotonic() - started), 0.001)
         lean_dir = ensure_package(project)
-        return lean_dir, build.build_modules(lean_dir, mods, PACKAGES, timeout, run)
+        return lean_dir, build.build_modules(lean_dir, mods, PACKAGES, remaining, run)
+    finally:
+        lock.release()
 
 
 def run_guarded(operation):

docker/lean/test_server_concurrency.py

@@ -99,26 +99,65 @@ def same_project_serialises(self, volume):
         self.assertEqual(len(probe.compiles), 2, "the later requests reuse what the first built")
 
     def test_a_restage_waits_for_the_build_in_progress(self):
+        # Ordered by events, not sleeps: the edit lands and the second request
+        # is on its way before the first compile reads its source back, and
+        # the compile gives the restage a generous window to run under it.
         project = self.project("p", volume=False)
         source = os.path.join(project, "lean", "Supermath", "Shared.lean")
+        compiling, requested, restaged = threading.Event(), threading.Event(), threading.Event()
         seen = []
+        stage = B.stage
+
+        def watched_stage(*args, **kw):
+            staged = stage(*args, **kw)
+            if compiling.is_set():
+                restaged.set()
+            return staged
 
         def run(argv, **kw):
             before = read(argv[-1])
-            time.sleep(0.15)
+            if not compiling.is_set():
+                compiling.set()
+                requested.wait(5)
+                restaged.wait(0.5)  # times out unless the restage runs under this compile
             seen.append(before == read(argv[-1]))
             return Probe(0)(argv, **kw)
 
         def edit_then_build():
-            time.sleep(0.05)
+            compiling.wait(5)
             write(source, "import Mathlib\n-- edited\n")
             os.utime(source, (2_000_000_000, 2_000_000_000))
+            requested.set()
             S.build_package(project, ["Supermath.Top"], 10, run)
 
-        errors = in_threads([lambda: S.build_package(project, ["Supermath.Top"], 10, run), edit_then_build])
+        B.stage = watched_stage
+        try:
+            errors = in_threads([lambda: S.build_package(project, ["Supermath.Top"], 10, run), edit_then_build])
+        finally:
+            B.stage = stage
         self.assertEqual(errors, [])
         self.assertTrue(seen and all(seen), "a source changed under a compile")
 
+    def test_a_request_waiting_for_the_lock_gives_up_at_its_timeout(self):
+        project = self.project("p", volume=True)
+        outcome = []
+
+        def request():
+            try:
+                S.build_package(project, ["Supermath.Top"], 0.3, Probe(0))
+                outcome.append("built")
+            except subprocess.TimeoutExpired:
+                outcome.append("timeout")
+
+        with B.project_lock(os.path.join(project, "lean")):
+            waiter = threading.Thread(target=request)
+            waiter.start()
+            waiter.join(3)
+            gave_up = not waiter.is_alive()
+        waiter.join(5)
+        self.assertTrue(gave_up, "the wait for the lock ignored the request's timeout")
+        self.assertEqual(outcome, ["timeout"])
+
     def test_checks_of_different_projects_build_at_the_same_time(self):
         first, second = self.project("a", True), self.project("b", False)
         both = threading.Barrier(2, timeout=5)