fleetd #409: deterministic test for the #399 completion-stamp ordering race

Widens the completed-hook's real race window (normally instructions-wide, needing
~2x-core host load to hit by chance per #399) by injecting a bounded sleep into the
test clock's completion-stamp read. This makes the ordering invariant — a test must
wait for isCompletionStampedForTest, not just DONE, before advancing the clock past
the TTL — fail deterministically on the first run when the barrier is removed, and
pass deterministically with it present. No production code changed.

fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java

@@ -1967,6 +1967,80 @@ class MessageServiceTest {
         }
     }
 
+    /**
+     * fleetd #409: pins the #399 ordering invariant deterministically, on the first run, without
+     * relying on host load.
+     *
+     * <p>fleetd #399 was itself only reproducible probabilistically: the real race window between
+     * {@code CompletableFuture.complete()} making {@link MessageService.Phase#DONE} observable and
+     * the constructor's {@code whenComplete} hook actually stamping {@code completedNanos} (see
+     * {@link MessageService#isCompletionStampedForTest}) is normally a handful of instructions wide,
+     * and needed heavy background load on the host to show up in a run at all. This test does not
+     * try to hit that narrow window by chance — it widens it on purpose: {@link #STAMP_DELAY_MILLIS}
+     * is injected into the clock itself, so the completion hook's one read of {@code nowNanos} for
+     * this ticket sleeps before returning, and everything the test does in the meantime (advance the
+     * clock, run a sweep, assert) happens for certain inside that window, on any host.
+     *
+     * <p>Removing the {@link #awaitCompletionStamped} call below reproduces the pre-#399 ordering:
+     * the test then advances the clock and runs its sweep while the hook is still asleep, so the
+     * hook wakes up and stamps {@code completedNanos} with the clock's ALREADY-ADVANCED value
+     * instead of the real completion time. {@code cutoff = advanced - TICKET_TTL_NANOS} can then
+     * never exceed that stamp (the gap between them is fixed at exactly {@code TICKET_TTL_NANOS}),
+     * so the sweep that already ran never evicts the ticket and the very next assertion — expecting
+     * eviction — fails immediately. That is a deterministic, first-run RED failure, not a flaky one
+     * and not a false pass: I ran the test with the barrier call removed and confirmed
+     * {@code assertNull} fails because {@code poll} still returns the ticket's DONE view, which is
+     * exactly this masked-eviction mechanism and not some unrelated defect in the test's own wiring.
+     */
+    @Test
+    void aTicketOrderedOnDoneInsteadOfTheCompletionStampSurvivesAnEvictionItMustNotSurvive() throws Exception {
+        java.util.concurrent.atomic.AtomicLong clock = new java.util.concurrent.atomic.AtomicLong(1_000_000_000L);
+        // Armed for exactly one call: MessageService's only three nowNanos() call sites are the Task
+        // constructor's createdNanos, this constructor's whenComplete hook's completedNanos, and
+        // pruneTerminalTickets' cutoff — none of which run between arming this (right before
+        // triggering the reply below) and the completion hook firing, so the delayed call is
+        // unambiguously that hook's stamp for `ticket`, never a createdNanos or cutoff read.
+        java.util.concurrent.atomic.AtomicBoolean delayArmed = new java.util.concurrent.atomic.AtomicBoolean(false);
+        java.util.function.LongSupplier gatedClock = () -> {
+            if (delayArmed.compareAndSet(true, false)) {
+                try {
+                    Thread.sleep(STAMP_DELAY_MILLIS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+            return clock.get();
+        };
+        MessageService service = new MessageService(agents, injector, rendezvous, inbox, null, null, gatedClock);
+        try {
+            String ticket = service.sendAsync(T, "a quick task");
+            awaitWaiting();
+            injectDelivery();
+
+            delayArmed.set(true);
+            assertTrue(rendezvous.resolve(T, "quick result"));
+
+            awaitTicketPhaseOn(service, ticket, MessageService.Phase.DONE);
+            // The barrier under test (fleetd #409, same fix as fleetd #399): remove this one call to
+            // reproduce the pre-#399 ordering — see the class-level note above for what happens then.
+            awaitCompletionStamped(service, ticket);
+
+            // A real clock would need the whole TTL to pass; the injected one does it instantly, and
+            // by now completedNanos already holds the REAL (small, unadvanced) completion time.
+            clock.addAndGet(MessageService.TICKET_TTL_NANOS + TimeUnit.SECONDS.toNanos(1));
+            service.sendAsync(T, "an unrelated second task"); // runs pruneTerminalTickets before returning
+
+            assertNull(service.poll(ticket),
+                    "a ticket whose real completion time is long past the advanced cutoff must be "
+                            + "evicted, whatever the completion hook's clock read was delayed by");
+        } finally {
+            service.close();
+        }
+    }
+
+    /** Bounded delay the injected clock sleeps for in {@link #aTicketOrderedOnDoneInsteadOfTheCompletionStampSurvivesAnEvictionItMustNotSurvive}. */
+    private static final long STAMP_DELAY_MILLIS = 300;
+
     private MessageService.TaskView awaitTicketPhaseOn(MessageService svc, String ticket,
             MessageService.Phase phase) throws Exception {
         long deadline = System.currentTimeMillis() + 3000;