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,9 @@ public void close() throws Exception
 
         if (config.drainOnClose())
         {
-            workers.forEach(EngineWorker::drain);
+            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,12 @@ public void doClose()
         }
         finally
         {
+            targetsByIndex.forEach((k, v) -> quietClose(v));
+            quietClose(streamsLayout);
+            quietClose(bufferPoolLayout);
+            debitorsByIndex.forEach((k, v) -> quietClose(v));
+            quietClose(creditor);
+            quietClose(eventWriter);
             thread = null;
         }
     }
@@ -1058,14 +1064,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)
         {