Bound event-loop termination waits and preserve close ownership

src/main/docs/bounded-termination.adoc

@@ -0,0 +1,46 @@
+= Bounded termination waits
+:sectnums:
+:lang: en-GB
+
+== Contract
+
+The thread that wins the transition to `STOPPING` owns the stop callback. Other
+callers wait at most five minutes of monotonic elapsed time. Expiry, interruption,
+a failed callback or reentrant stop throws `IllegalStateException`; interruption
+retains the interrupted flag. Only a successful callback sets `STOPPED`.
+If the callback completes between a waiter's initial state read and its expiry
+or interruption check, completed shutdown takes precedence; the interrupted flag
+is still preserved. An unfinished callback retains the failure behaviour.
+
+The first failed wait records the loop name, state, elapsed time, stopper stack
+and up to eight owned event-loop thread stacks, limited to 64 frames each.
+Subsequent failures still throw but do not repeat the diagnostic. The timeout
+bounds secondary waits; it cannot forcibly complete an application stop callback.
+
+`close()` performs this stop check before entering `AbstractCloseable`'s close
+machinery, which otherwise catches exceptions and marks the object closed.
+An unsuccessful stop therefore leaves the close state and handlers untouched.
+After a blocked callback eventually finishes, the caller can close normally.
+
+The clock, deadline, failure/owner bookkeeping, handler-finished flags and pending
+handler admission list are transient runtime state. The pending list is a subset
+of the existing handler collection, rather than separate configuration.
+They are excluded from reflective Wire configuration marshalling, preserving the
+existing event-group representation. Chronicle-Wire's `MarshallingEventGroupTest`
+provides the downstream compatibility control.
+
+== Regression coverage
+
+`TerminationWaitTest` controls elapsed time independently of wall-clock changes
+and exercises normal/repeated stop, concurrent callers, expiry through stop and
+close, interruption, callback failure, one diagnostic and retained ownership.
+All controlled worker threads are released and joined after the assertion.
+`completedStopWinsRaceWithWaitFailure` completes the real stop callback during
+the controlled clock read, covering both interruption and deadline expiry.
+Stack assertions identify the owned stop callback without requiring it to have
+reached a particular instruction after releasing the observer's latch.
+
+The surviving-runner lookup fixture keeps both handlers live for the lookup
+assertions, then waits for their actual removal before closing the loop. Merely
+releasing their latches did not guarantee that teardown could no longer interrupt
+the second handler's wait.

src/main/docs/decision-log.adoc

@@ -246,6 +246,16 @@ Impact & Consequences::
 * Explicitly recorded, bounded repeat runs remain useful for diagnosis.
 * CI must still handle reported failures when `maven.test.failure.ignore` allows later build steps to continue.
 
+[[THR-NF-O-038]]
+=== THR-NF-O-038 Bound unsuccessful termination waits
+
+Date:: 2026-09-18
+Context:: A shutdown deadline logged on every poll without ending the wait; close also swallowed stop failures and could release live resources.
+Decision Statement:: Use monotonic elapsed time, throw on unsuccessful waits, emit one bounded diagnostic and stop before committing the close state.
+Impact & Consequences:: A failed wait retains resource ownership and does not claim completion. The original stop callback remains responsible for finishing. Interrupted callers keep their interrupted status.
+Validation:: `TerminationWaitTest` covers timeout through stop and close, interruption, callback failure, normal and concurrent stop, and eventual cleanup.
+Notes/Links:: link:bounded-termination.adoc[Full termination contract and diagnostic limits].
+
 [[THR-FN-037]]
 === THR-FN-037 Preserve legacy registration with explicit checked admission
 

src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java

@@ -12,7 +12,9 @@
 import org.jetbrains.annotations.NotNull;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.LongSupplier;
 
 /**
  * Base implementation that manages the life-cycle of an {@link EventLoop}.
@@ -30,17 +32,25 @@
  * Transitions are linear in that order. Invoking {@code stop()} while in
  * {@code NEW} skips {@code STARTED} entirely. Both {@code start()} and
  * {@code stop()} are idempotent and {@code stop()} blocks until the loop is
- * {@code STOPPED}.
+ * {@code STOPPED}. A failed or interrupted termination wait throws without
+ * claiming that shutdown completed.
  */
 @SuppressWarnings("this-escape")
 public abstract class AbstractLifecycleEventLoop extends AbstractCloseable implements EventLoop {
 
     /**
-     * After this time, awaitTermination will log an error and return, this is really only so
-     * tests don't block forever. This time should be kept as "effectively forever".
+     * Bound a secondary caller's wait for the thread already stopping the loop.
      */
     private static final long AWAIT_TERMINATION_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(5);
     private final AtomicReference<EventLoopLifecycle> lifecycle = new AtomicReference<>(EventLoopLifecycle.NEW);
+    //! Termination bookkeeping belongs to the live loop, not its marshalled configuration.
+    //! In particular a clock lambda and a stopping thread cannot be portable wire data.
+    //! Integration control: Chronicle-Wire's MarshallingEventGroupTest.test.
+    private transient final AtomicBoolean terminationFailureReported = new AtomicBoolean();
+    private transient final long terminationTimeoutNs;
+    private transient final LongSupplier nanoClock;
+    private transient volatile Thread stoppingThread;
+    private transient volatile Throwable stopFailure;
     protected final String name;
     volatile boolean privateGroup;
 
@@ -54,6 +64,15 @@
      * @param name descriptive name for the loop
      */
     protected AbstractLifecycleEventLoop(@NotNull String name) {
+        this(name, TimeUnit.MILLISECONDS.toNanos(AWAIT_TERMINATION_TIMEOUT_MS), System::nanoTime);
+    }
+
+    // Package-local seam: tests advance elapsed time without changing the production deadline.
+    AbstractLifecycleEventLoop(@NotNull String name, long terminationTimeoutNs, LongSupplier nanoClock) {
+        if (terminationTimeoutNs <= 0)
+            throw new IllegalArgumentException("Termination timeout must be positive");
+        this.terminationTimeoutNs = terminationTimeoutNs;
+        this.nanoClock = nanoClock;
         this.name = name.replaceAll("/$", "");
 
         // event loops operate on dedicated threads but may be closed elsewhere
@@ -120,16 +139,30 @@
     @Override
     public final void stop() {
         if (lifecycle.compareAndSet(EventLoopLifecycle.NEW, EventLoopLifecycle.STOPPING)) {
-            performStopFromNew();
-            lifecycle.set(EventLoopLifecycle.STOPPED);
+            performStop(false);
         } else if (lifecycle.compareAndSet(EventLoopLifecycle.STARTED, EventLoopLifecycle.STOPPING)) {
-            performStopFromStarted();
-            lifecycle.set(EventLoopLifecycle.STOPPED);
+            performStop(true);
         } else {
             awaitTermination();
         }
     }
 
+    private void performStop(boolean started) {
+        stoppingThread = Thread.currentThread();
+        try {
+            if (started)
+                performStopFromStarted();
+            else
+                performStopFromNew();
+            lifecycle.set(EventLoopLifecycle.STOPPED);
+        } catch (RuntimeException | Error failure) {
+            stopFailure = failure;
+            throw failure;
+        } finally {
+            stoppingThread = null;
+        }
+    }
+
     /**
      * Stop the loop when {@link #stop()} is invoked before it has started.
      * Implementations should block until every handler has received
@@ -149,22 +182,67 @@
      *
      * <p>If the state does not change within
      * {@link #AWAIT_TERMINATION_TIMEOUT_MS} milliseconds an error is logged and
-     * the method returns. The timeout is primarily to avoid tests hanging
-     * indefinitely.</p>
+     * an {@link IllegalStateException} is thrown. Interruption preserves the
+     * interrupted status and also fails the wait. Neither case completes the
+     * lifecycle or transfers ownership of resources.</p>
      */
     protected final void awaitTermination() {
-        long endTime = System.currentTimeMillis() + AWAIT_TERMINATION_TIMEOUT_MS;
-        while (!Thread.currentThread().isInterrupted()) {
+        long start = nanoClock.getAsLong();
+        while (true) {
             if (lifecycle.get() == EventLoopLifecycle.STOPPED)
                 return;
-            if (System.currentTimeMillis() > endTime) {
-                Jvm.error().on(getClass(), "awaitTermination() timed out, continuing. This probably represents a bug.");
+            long elapsed = nanoClock.getAsLong() - start;
+            if (stopFailure != null)
+                throw terminationFailure("stop callback failed", elapsed);
+            if (stoppingThread == Thread.currentThread())
+                throw terminationFailure("reentrant stop", elapsed);
+            if (Thread.currentThread().isInterrupted() || elapsed >= terminationTimeoutNs) {
+                //! Shutdown can complete after the loop's first state read, including while
+                //! interrupting its workers. Honour completed ownership transfer before failing.
+                //! Control: TerminationWaitTest.completedStopWinsRaceWithWaitFailure.
+                if (lifecycle.get() == EventLoopLifecycle.STOPPED)
+                    return;
+                throw terminationFailure(Thread.currentThread().isInterrupted() ? "interrupted" : "timed out", elapsed);
             }
             Jvm.pause(1);
         }
-        if (lifecycle.get() != EventLoopLifecycle.STOPPED) {
-            Jvm.warn().on(getClass(), "awaitTermination() interrupted, returning in state " + lifecycle.get());
+    }
+
+    private IllegalStateException terminationFailure(String reason, long elapsedNs) {
+        StringBuilder diagnostic = new StringBuilder("awaitTermination() ").append(reason)
+                .append(": loop=").append(name).append(", lifecycle=").append(lifecycle.get())
+                .append(", elapsedMs=").append(TimeUnit.NANOSECONDS.toMillis(elapsedNs));
+        Thread stopper = stoppingThread;
+        appendThread(diagnostic, "stopper", stopper);
+        int remaining = 8;
+        for (Thread thread : Thread.getAllStackTraces().keySet()) {
+            if (thread != stopper && isRunningOnThread(thread)) {
+                appendThread(diagnostic, "event loop", thread);
+                if (--remaining == 0) {
+                    diagnostic.append("\nFurther event-loop threads omitted");
+                    break;
+                }
+            }
+        }
+        IllegalStateException failure = new IllegalStateException(diagnostic.toString(), stopFailure);
+        // A timeout must not produce one error per millisecond, or per subsequent close call.
+        if (terminationFailureReported.compareAndSet(false, true))
+            Jvm.error().on(getClass(), diagnostic.toString(), failure);
+        return failure;
+    }
+
+    @SuppressWarnings("deprecation") // Thread.threadId() is unavailable on the supported Java 8 baseline.
+    private static void appendThread(StringBuilder diagnostic, String role, Thread thread) {
+        diagnostic.append('\n').append(role).append('=');
+        if (thread == null) {
+            diagnostic.append("none");
+            return;
         }
+        diagnostic.append(thread.getName()).append(" id=").append(thread.getId())
+                .append(" state=").append(thread.getState());
+        StackTraceElement[] stack = thread.getStackTrace();
+        for (int i = 0; i < Math.min(stack.length, 64); i++)
+            diagnostic.append("\n  at ").append(stack[i]);
     }
 
     @Override
@@ -177,6 +255,9 @@
         if (!privateGroup && isRunningOnThread(Thread.currentThread())) {
             throw new ThreadingIllegalStateException(getClass() + ": Attempting to close " + name + " from within!", createdHere());
         }
+        // AbstractCloseable swallows performClose failures and marks the object closed.
+        // Stop before entering that path so failure cannot release a live loop's handlers.
+        stop();
     }
 
     public abstract boolean isRunningOnThread(Thread thread);

src/main/java/net/openhft/chronicle/threads/BlockingEventLoop.java

@@ -43,7 +43,9 @@
     @NotNull
     private transient final ExecutorService service;
     private final List<EventHandler> handlers = new CopyOnWriteArrayList<>();
-    private final List<EventHandler> pendingHandlers = new ArrayList<>();
+    //! Pending admission tracks a runtime subset of handlers, which are already represented above.
+    //! Keep this ownership bookkeeping out of Wire: MarshallingEventGroupTest.test in Chronicle-Wire.
+    private transient final List<EventHandler> pendingHandlers = new ArrayList<>();
     private final List<Runner> runners = new CopyOnWriteArrayList<>();
     private final NamedThreadFactory threadFactory;
     private final Supplier<Pauser> pauserSupplier;

src/main/java/net/openhft/chronicle/threads/MediumEventLoop.java

@@ -47,7 +47,9 @@ public class MediumEventLoop extends AbstractLifecycleEventLoop implements CoreE
      */
     private final transient Object addHandlerMutex = new Object();
     private final transient Object startStopMutex = new Object();
-    private volatile boolean handlersFinished;
+    //! Shutdown callback bookkeeping is runtime state, not Wire configuration.
+    //! Compatibility control: Chronicle-Wire's MarshallingEventGroupTest.test.
+    private transient volatile boolean handlersFinished;
 
     @Nullable
     protected final transient EventLoop parent;

src/main/java/net/openhft/chronicle/threads/MonitorEventLoop.java

@@ -42,7 +42,9 @@
     private final List<EventHandler> handlers = new CopyOnWriteArrayList<>();
     private final Pauser pauser;
     private transient volatile Thread thread = null;
-    private boolean handlersFinished;
+    //! Shutdown callback bookkeeping is runtime state, not Wire configuration.
+    //! Compatibility control: Chronicle-Wire's MarshallingEventGroupTest.test.
+    private transient boolean handlersFinished;
 
     public MonitorEventLoop(final EventLoop parent, final Pauser pauser) {
         this(parent, "", pauser);

src/test/java/net/openhft/chronicle/threads/BlockingEventLoopShutdownTest.java

@@ -68,7 +68,7 @@ void lookupFindsSurvivingRunnerWhenEarlierRunnerIsRemoved() throws IllegalAccess
         final CountDownLatch finishFirst = new CountDownLatch(1);
         final CountDownLatch finishSecond = new CountDownLatch(1);
         final AtomicReference<Thread> survivor = new AtomicReference<>();
-        final RemovingRunnerList runners = new RemovingRunnerList(finishFirst);
+        final RemovingRunnerList runners = new RemovingRunnerList(finishFirst, 2);
         runners.finishAfterGet = true;
         try (BlockingEventLoop loop = new BlockingEventLoop("surviving-runner")) {
             Jvm.getField(BlockingEventLoop.class, "runners").set(loop, runners);
@@ -95,6 +95,10 @@ void lookupFindsSurvivingRunnerWhenEarlierRunnerIsRemoved() throws IllegalAccess
             } finally {
                 finishFirst.countDown();
                 finishSecond.countDown();
+                // Releasing the latch only makes the survivor runnable. Closing immediately
+                // can interrupt await before it returns and turn normal teardown into a WARN.
+                // Keep the live-runner assertions above, then await both actual removals.
+                await(runners.allRemoved);
             }
         }
     }
@@ -112,12 +116,18 @@ private static final class RemovingRunnerList extends CopyOnWriteArrayList<Objec
         private static final long serialVersionUID = 1L;
         private final transient CountDownLatch finish;
         private final transient CountDownLatch removed = new CountDownLatch(1);
+        private final transient CountDownLatch allRemoved;
         private final AtomicBoolean armed = new AtomicBoolean();
         private final AtomicInteger iterators = new AtomicInteger();
         private boolean finishAfterGet;
 
         private RemovingRunnerList(CountDownLatch finish) {
+            this(finish, 1);
+        }
+
+        private RemovingRunnerList(CountDownLatch finish, int expectedRemovals) {
             this.finish = finish;
+            this.allRemoved = new CountDownLatch(expectedRemovals);
         }
 
         @Override
@@ -154,8 +164,10 @@ private void finishRunner() {
         @Override
         public boolean remove(Object runner) {
             final boolean result = super.remove(runner);
-            if (result)
+            if (result) {
                 removed.countDown();
+                allRemoved.countDown();
+            }
             return result;
         }
     }