fix(engine): release worker buffers in coordinated close to avoid JVM crash on binding fault

A hot-path binding exception is wrapped by EngineWorker.doWork as AgentTerminationException, terminating the worker agent. Agrona then runs onClose on the agent thread, which released the worker's memory-mapped buffers in advance. The engine-coordinated Engine.close() -> drain() then read that worker's unmapped streams ring buffer via Unsafe, causing a SIGSEGV / JVM abort.

Defer all worker memory-mapped resource releases (targets, streamsLayout, bufferPoolLayout, debitors, creditor, eventWriter) from onClose to doClose's finally, which runs after the agent has stopped, so a self-terminated worker no longer unmaps buffers the engine or peer workers still reference. Engine.close also skips draining workers whose agent already terminated, which would otherwise spin until the drain timeout.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jfallows

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/Engine.java

@@ -332,7 +332,11 @@ public void close() throws Exception
 
         if (config.drainOnClose())
         {
-            workers.forEach(EngineWorker::drain);
+            // skip workers whose agent already terminated (e.g. on a binding fault); they will
+            // never consume again, so draining them would just spin until the timeout
+            workers.stream()
+                   .filter(worker -> !worker.runner().isClosed())
+                   .forEach(EngineWorker::drain);
         }
 
         final List<Throwable> errors = new ArrayList<>();

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/registry/EngineWorker.java

@@ -881,6 +881,14 @@ public void doClose()
         }
         finally
         {
+            // release memory-mapped resources only after the agent has stopped, not in onClose,
+            // so a self-terminated worker does not unmap buffers the engine or peers still reference
+            targetsByIndex.forEach((k, v) -> quietClose(v));
+            quietClose(streamsLayout);
+            quietClose(bufferPoolLayout);
+            debitorsByIndex.forEach((k, v) -> quietClose(v));
+            quietClose(creditor);
+            quietClose(eventWriter);
             thread = null;
         }
     }
@@ -1058,14 +1066,6 @@ public void onClose()
         }
 
         targetsByIndex.forEach((k, v) -> v.detach());
-        targetsByIndex.forEach((k, v) -> quietClose(v));
-
-        quietClose(streamsLayout);
-        quietClose(bufferPoolLayout);
-
-        debitorsByIndex.forEach((k, v) -> quietClose(v));
-        quietClose(creditor);
-        quietClose(eventWriter);
 
         if (acquiredBuffers != 0 || acquiredCreditors != 0 || acquiredDebitors != 0L)
         {