test(pool): prevent failed-stop eviction regression deadlock

pulse/pool/tla/eviction_test/README.md

@@ -0,0 +1,51 @@
+# Failed-stop eviction test ordering
+
+`StopEvictionTest.tla` models the callback gate in
+`TestEvictionStopsJobRestoredByFailedStop`, rather than changing the worker's
+runtime contract. The failed main CI run
+[38646089002](https://github.com/CaliLuke/loom/actions/runs/38646089002) shows
+the event loop holding the gated callback, the test blocked in `Worker.stop`,
+and the direct `stopJob` caller waiting for the transition lock.
+
+## Decision and correspondence
+
+- `Publish`: `stopJob` calls `issueStop`, which publishes a stream event before
+  processing the returned operation directly. Either the direct caller or
+  `handleEvents` can reach `handler.Stop` first (`Enter`).
+- `Join`: `Worker.stop` joins the event loop. In the original test this call
+  preceded `close(handler.release)`, so a runtime-owned callback prevented the
+  test from ever releasing its own gate. `current.cfg` reproduces that cycle
+  with a `NoJoinCycle` violation (5 generated / 4 distinct states, exit 12).
+- `Quiesce`: the corrected test joins the runtime before issuing the gated
+  stop. Only the direct caller can now enter the callback. The later join is
+  vacuous because no runtime goroutine remains.
+- `Release` / `FailStop` / `Evict`: eviction waits for the failed callback;
+  the job remains tracked on failure and eviction retries Stop and forgets it.
+  The Go test checks the handler error, retained job while blocked, eviction
+  completion, empty final jobs, and exactly two Stop calls.
+
+`quiesced.cfg` passes all three invariants (8 generated / 8 distinct states,
+exit 0). This bounded model assumes finite runtime shutdown before publication,
+successful eviction's second callback, and no other ownership transition. It
+does not prove Redis delivery, lease timing, general worker shutdown liveness,
+or mutex implementation correctness. The runtime behavior and production locks
+are unchanged. Repeated and race-enabled Go tests check the actual test harness.
+
+## Run
+
+Use TLC 2.19 from `tlaplus/tlaplus` release `v1.7.4` (`tla2tools.jar`, SHA256
+`936a262061c914694dfd669a543be24573c45d5aa0ff20a8b96b23d01e050e88`), Java 25:
+
+```sh
+cd pulse/pool/tla/eviction_test
+java -XX:+UseParallelGC -cp /path/to/tla2tools.jar tlc2.TLC \
+  -workers 1 -deadlock -metadir /tmp/loom-eviction-current-states \
+  -config current.cfg StopEvictionTest.tla
+java -XX:+UseParallelGC -cp /path/to/tla2tools.jar tlc2.TLC \
+  -workers 1 -deadlock -metadir /tmp/loom-eviction-quiesced-states \
+  -config quiesced.cfg StopEvictionTest.tla
+```
+
+`-deadlock` allows the normal terminal state; `NoJoinCycle` identifies the
+invalid circular wait. Keep checker traces and state directories outside the
+repository and remove task-owned output after review.

pulse/pool/tla/eviction_test/StopEvictionTest.tla

@@ -0,0 +1,38 @@
+------------------------- MODULE StopEvictionTest -------------------------
+EXTENDS Naturals, TLC
+CONSTANT QuiesceFirst
+VARIABLES phase, runtime, callback, released, tracked, evicted
+vars == <<phase, runtime, callback, released, tracked, evicted>>
+
+Init == /\ phase = IF QuiesceFirst THEN "quiesce" ELSE "publish"
+        /\ runtime = TRUE /\ callback = "none" /\ released = FALSE
+        /\ tracked = TRUE /\ evicted = FALSE
+
+Quiesce == /\ phase = "quiesce"
+           /\ runtime' = FALSE /\ phase' = "publish"
+           /\ UNCHANGED <<callback, released, tracked, evicted>>
+Publish == /\ phase = "publish" /\ phase' = "callback"
+           /\ UNCHANGED <<runtime, callback, released, tracked, evicted>>
+Enter(owner) == /\ phase = "callback" /\ callback = "none"
+                /\ (owner = "direct" \/ runtime)
+                /\ callback' = owner /\ phase' = "join"
+                /\ UNCHANGED <<runtime, released, tracked, evicted>>
+\* Worker.stop joins the runtime. A callback on that runtime must return first.
+Join == /\ phase = "join" /\ callback # "runtime"
+        /\ runtime' = FALSE /\ phase' = "release"
+        /\ UNCHANGED <<callback, released, tracked, evicted>>
+Release == /\ phase = "release" /\ released' = TRUE /\ phase' = "evict"
+           /\ UNCHANGED <<runtime, callback, tracked, evicted>>
+FailStop == /\ released /\ callback \in {"direct", "runtime"}
+            /\ callback' = "failed"
+            /\ UNCHANGED <<phase, runtime, released, tracked, evicted>>
+Evict == /\ phase = "evict" /\ callback = "failed"
+         /\ tracked' = FALSE /\ evicted' = TRUE /\ phase' = "done"
+         /\ UNCHANGED <<runtime, callback, released>>
+Next == Quiesce \/ Publish \/ Enter("direct") \/ Enter("runtime")
+        \/ Join \/ Release \/ FailStop \/ Evict
+Spec == Init /\ [][Next]_vars
+NoJoinCycle == ~(phase = "join" /\ callback = "runtime" /\ ~released)
+FailedStopRetainsJob == callback = "failed" /\ ~evicted => tracked
+EvictionForgetsJob == evicted => ~tracked /\ callback = "failed"
+=============================================================================

pulse/pool/tla/eviction_test/current.cfg

@@ -0,0 +1,3 @@
+CONSTANT QuiesceFirst = FALSE
+SPECIFICATION Spec
+INVARIANTS NoJoinCycle FailedStopRetainsJob EvictionForgetsJob

pulse/pool/tla/eviction_test/quiesced.cfg

@@ -0,0 +1,3 @@
+CONSTANT QuiesceFirst = TRUE
+SPECIFICATION Spec
+INVARIANTS NoJoinCycle FailedStopRetainsJob EvictionForgetsJob

pulse/pool/worker_evict_test.go

@@ -330,31 +330,50 @@ func waitBriefly(done <-chan struct{}) {
 }
 
 // TestEvictionStopsJobRestoredByFailedStop covers an eviction that overlaps a
-// failing stopJob, as in shutdown's stopAllJobs: stopJob takes k, eviction
-// starts, then Stop fails and stopJob tracks k again. Eviction must still
+// failing stopJob, as in shutdown's stopAllJobs: Stop holds k, eviction
+// starts, then Stop fails and stopJob leaves k tracked. Eviction must still
 // stop k and leave the worker with no tracked job.
 func TestEvictionStopsJobRestoredByFailedStop(t *testing.T) {
 	handler := &failingStopHandler{failFirst: true, entered: make(chan struct{}), release: make(chan struct{})}
 	_, worker := startBlockingStopWorker(t, "evict-failed-stop", "k1", handler)
-	stopDone := make(chan struct{})
+	release := sync.OnceFunc(func() {
+		close(handler.release)
+	})
+	defer release()
+	// stopJob publishes a stop event as well as processing it directly. Join
+	// the runtime before gating Stop, so the event loop cannot take the
+	// callback and make worker.stop wait for a release owned by this test.
+	worker.stop(t.Context())
+	stopDone := make(chan error, 1)
 	go func() {
-		defer close(stopDone)
-		if err := worker.stopJob(t.Context(), "k1"); err == nil {
-			t.Errorf("stopJob succeeded, want the handler error")
-		}
+		stopDone <- worker.stopJob(t.Context(), "k1")
 	}()
 	waitEntered(t, handler)
 
-	worker.stop(t.Context())
 	evictDone := make(chan struct{})
 	go func() {
 		worker.stopHandlers()
 		close(evictDone)
 	}()
 	waitBriefly(evictDone)
-	close(handler.release)
-	<-stopDone
-	<-evictDone
+	select {
+	case <-evictDone:
+		t.Error("eviction returned before the in-progress Stop finished")
+	default:
+	}
+	require.Len(t, worker.Jobs(), 1, "in-progress Stop must retain the tracked job")
+	release()
+	select {
+	case err := <-stopDone:
+		require.ErrorContains(t, err, "stop failed")
+	case <-time.After(5 * time.Second):
+		t.Fatal("stopJob did not return after releasing the handler")
+	}
+	select {
+	case <-evictDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("eviction did not finish after the failed Stop")
+	}
 
 	require.Empty(t, worker.Jobs(), "evicted worker still tracks a job")
 	require.Equal(t, int32(2), handler.stops.Load(), "handler.Stop calls")