Separate engine initialization from startup (#1808)

* feat(engine): split Engine.start() into init() + start() with beforeStart hook

Introduce Engine.init() to start workers and boss without attaching
bindings; it is idempotent. Engine.start() now calls init() implicitly
before reading config and attaching bindings, so existing callers (CLI
start command, embedded users) are unaffected.

EngineRule gains a beforeStart(Runnable) builder (default no-op). The rule
now calls engine.init(), then runs the beforeStart hook and engine.start()
on a separate thread, allowing tests to block binding startup against an
external initialization signal (e.g. k3po awaitPrepared) before bindings
attach.

Closes #1807

https://claude.ai/code/session_01XcNeWBUJAA87NSf9R4KdVT

* refactor(engine): only spawn EngineRule start thread when beforeStart is set

Keep beforeStart null by default and branch on it: when no hook is set,
EngineRule starts the engine synchronously on the test thread exactly as
before, with no background thread. The init() + background start() path is
used only when a beforeStart hook is provided, preserving existing behavior
precisely for the common case.

https://claude.ai/code/session_01XcNeWBUJAA87NSf9R4KdVT

* refactor(engine): drop redundant beforeStart null assignment in EngineRule

Address review feedback: remove the explicit null initialization since
null is the implicit default for the reference field.

https://claude.ai/code/session_01XcNeWBUJAA87NSf9R4KdVT

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: jfallows

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

@@ -105,6 +105,7 @@ public final class Engine implements Collector, AutoCloseable
     private final RouterConfig routerConfig;
 
     private final EventWriter eventWriter;
+    private final AtomicBoolean initialized;
     private final AtomicBoolean closed;
 
     private FileSystem fileSystem = null;
@@ -271,6 +272,7 @@ public final class Engine implements Collector, AutoCloseable
         this.readonly = readonly;
         this.manager = manager;
         this.diagnostics = diagnostics;
+        this.initialized = new AtomicBoolean(false);
         this.closed = new AtomicBoolean(false);
     }
 
@@ -290,14 +292,23 @@ private void process(
         manager.process(config);
     }
 
-    public void start() throws Exception
+    public void init()
     {
-        for (EngineWorker worker : workers)
+        if (initialized.compareAndSet(false, true))
         {
-            worker.doStart();
+            for (EngineWorker worker : workers)
+            {
+                worker.doStart();
+            }
+
+            boss.doStart();
         }
+    }
 
-        boss.doStart();
+    public void start() throws Exception
+    {
+        // start workers and boss if init() has not already been called; idempotent
+        init();
 
         // ignore the config file in read-only mode; no config will be read so no namespaces, bindings, etc. will be attached
         if (!readonly)

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/internal/EngineTest.java

@@ -74,6 +74,53 @@ public void shouldConfigureEmpty()
         }
     }
 
+    @Test
+    public void shouldInitThenStart()
+    {
+        EngineConfiguration config = new EngineConfiguration(properties);
+        List<Throwable> errors = new LinkedList<>();
+        try (Engine engine = Engine.builder()
+                .config(config)
+                .errorHandler(errors::add)
+                .build())
+        {
+            engine.init();
+            engine.start();
+        }
+        catch (Throwable ex)
+        {
+            errors.add(ex);
+        }
+        finally
+        {
+            assertThat(errors, empty());
+        }
+    }
+
+    @Test
+    public void shouldInitIdempotent()
+    {
+        EngineConfiguration config = new EngineConfiguration(properties);
+        List<Throwable> errors = new LinkedList<>();
+        try (Engine engine = Engine.builder()
+                .config(config)
+                .errorHandler(errors::add)
+                .build())
+        {
+            engine.init();
+            engine.init();
+            engine.start();
+        }
+        catch (Throwable ex)
+        {
+            errors.add(ex);
+        }
+        finally
+        {
+            assertThat(errors, empty());
+        }
+    }
+
     @Test
     public void shouldConfigure()
     {

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/EngineRule.java

@@ -93,6 +93,7 @@ public final class EngineRule implements TestRule
     private Predicate<String> exceptions;
     private boolean interruptible;
     private boolean clean;
+    private Runnable beforeStart;
 
     public EngineRule()
     {
@@ -180,6 +181,13 @@ public EngineRule clean()
         return this;
     }
 
+    public EngineRule beforeStart(
+        Runnable beforeStart)
+    {
+        this.beforeStart = requireNonNull(beforeStart);
+        return this;
+    }
+
     public void close()
     {
         try
@@ -391,9 +399,49 @@ public void evaluate() throws Throwable
                                 .errorHandler(errorHandler)
                                 .build();
 
+                // when no beforeStart hook is set, preserve existing behavior exactly: start (and implicitly
+                // init) the engine synchronously on the test thread before the test body runs.
+                // when a hook is set, init the engine eagerly and run the hook followed by start() on a
+                // separate thread so the hook can block (e.g. waiting for an external runtime to be ready)
+                // without stalling the test body, while bindings attach once the hook returns.
+                final Thread starter;
+                if (beforeStart == null)
+                {
+                    starter = null;
+                }
+                else
+                {
+                    engine.init();
+                    starter = new Thread(() ->
+                    {
+                        try
+                        {
+                            beforeStart.run();
+                            engine.start();
+                        }
+                        catch (Throwable t)
+                        {
+                            errors.add(t);
+
+                            if (interruptible)
+                            {
+                                baseThread.interrupt();
+                            }
+                        }
+                    });
+                    starter.setName("engine-rule-before-start");
+                }
+
                 try
                 {
-                    engine.start();
+                    if (starter != null)
+                    {
+                        starter.start();
+                    }
+                    else
+                    {
+                        engine.start();
+                    }
 
                     base.evaluate();
                 }
@@ -405,6 +453,10 @@ public void evaluate() throws Throwable
                 {
                     try
                     {
+                        if (starter != null)
+                        {
+                            starter.join();
+                        }
                         engine.close();
                     }
                     catch (Throwable t)