REPL.3: eager extension init off the prompt thread + init observability

build.sbt

@@ -20,7 +20,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "softclient4es"
 
-ThisBuild / version := "0.20.1"
+ThisBuild / version := "0.20.2-SNAPSHOT"
 
 ThisBuild / scalaVersion := scala213
 

core/src/main/scala/app/softnetwork/elastic/client/Cli.scala

@@ -24,6 +24,8 @@ import scala.concurrent.ExecutionContext
 
 object Cli extends App {
 
+  private val logger = org.slf4j.LoggerFactory.getLogger(getClass)
+
   implicit val system: ActorSystem = ActorSystem("softclient4es-sql-cli")
   implicit val ec: ExecutionContext = system.dispatcher
 
@@ -33,6 +35,27 @@ object Cli extends App {
   try {
     val gateway = ElasticClientFactory.createWithMonitoring(config.elasticConfig)
 
+    // #163 — eager extension init OFF the prompt thread: fold the ~442ms ServiceLoader scan (and
+    // kick off any extension-side warm-up, e.g. the arrow JoinExtension's DuckDB native load)
+    // into startup instead of the first query. Fire-and-forget: the REPL prompt (or the batch
+    // statement) proceeds immediately; a query racing this simply blocks on the lazy-val monitor
+    // (single scan, no double-init). Goes through the delegator forwarding so the DELEGATE's
+    // registry — the one run() consults — is the one warmed.
+    locally {
+      val initStart = System.nanoTime()
+      gateway.initializeExtensions().onComplete {
+        case scala.util.Success(count) =>
+          logger.info(
+            s"🔌 $count extension(s) ready in ${(System.nanoTime() - initStart) / 1000000L}ms"
+          )
+        case scala.util.Failure(e) =>
+          logger.warn(
+            "⚠️ Eager extension initialization failed — extensions will initialize on first query",
+            e
+          )
+      }
+    }
+
     val executor = new StreamingReplExecutor(gateway)
     val repl = new Repl(executor, config.replConfig)
 

core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala

@@ -1907,6 +1907,15 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
   ): Future[ElasticResult[QueryResult]] =
     delegate.run(statement)
 
+  // ==================== Extensions (delegate) ====================
+
+  /** #163 — the delegator inherits ExtensionApi's lazy vals, so `this.extensionRegistry` would be a
+    * SECOND registry that no query path consults (`run` delegates to `delegate.run`, which uses the
+    * delegate's registry). Eager init must warm THAT one.
+    */
+  override def initializeExtensions()(implicit ec: ExecutionContext): Future[Int] =
+    delegate.initializeExtensions()
+
   // ==================== Transform (delegate) ====================
 
   override def createTransform(

core/src/main/scala/app/softnetwork/elastic/client/ExtensionApi.scala

@@ -23,6 +23,8 @@ import app.softnetwork.elastic.licensing.{
   LicenseRefreshStrategyFactory
 }
 
+import scala.concurrent.{ExecutionContext, Future}
+
 trait ExtensionApi { self: ElasticClientApi =>
 
   /** License refresh strategy resolved via LicenseRefreshStrategyFactory. The factory uses SPI
@@ -55,4 +57,16 @@ trait ExtensionApi { self: ElasticClientApi =>
   /** Extension registry (lazy loaded) */
   lazy val extensionRegistry: ExtensionRegistry =
     new ExtensionRegistry(config, licenseRefreshStrategy)
+
+  /** #163 — eagerly force extension discovery + initialization on the given execution context,
+    * instead of paying the ServiceLoader scan (~442ms measured over a 278-jar REPL classpath) on
+    * the first query. Additive: the underlying lazy vals stay the single initialization path — a
+    * concurrent first query simply blocks on the lazy-val monitor until the one scan finishes (and
+    * a FAILED lazy init is not cached, so a later query retries).
+    *
+    * @return
+    *   the number of successfully initialized extensions.
+    */
+  def initializeExtensions()(implicit ec: ExecutionContext): Future[Int] =
+    Future(extensionRegistry.extensions.size)
 }

core/src/main/scala/app/softnetwork/elastic/client/ExtensionRegistry.scala

@@ -35,28 +35,41 @@ class ExtensionRegistry(
   /** All loaded extensions.
     */
   lazy val extensions: Seq[ExtensionSpi] = {
+    val scanStart = System.nanoTime()
     val loader = ServiceLoader.load(classOf[ExtensionSpi])
 
     val loaded = loader.iterator().asScala.toSeq.flatMap { ext =>
       logger.info(
         s"🔌 Discovered extension: ${ext.extensionName} v${ext.version} (priority: ${ext.priority})"
       )
 
+      val initStart = System.nanoTime()
       ext.initialize(config, licenseRefreshStrategy) match {
         case Right(_) =>
-          logger.info(s"✅ Extension ${ext.extensionName} initialized successfully")
+          logger.info(
+            s"✅ Extension ${ext.extensionName} initialized successfully in ${elapsedMs(initStart)}ms"
+          )
           Some(ext)
 
         case Left(error) =>
-          logger.warn(s"⚠️ Failed to initialize extension ${ext.extensionName}: $error")
+          logger.warn(
+            s"⚠️ Failed to initialize extension ${ext.extensionName} after ${elapsedMs(initStart)}ms: $error"
+          )
           None
       }
     }
 
     // ✅ Sort by priority (lower = higher priority)
-    loaded.sortBy(_.priority)
+    val sorted = loaded.sortBy(_.priority)
+    logger.info(
+      s"🔌 Extension discovery + initialization completed in ${elapsedMs(scanStart)}ms " +
+      s"(${sorted.size} extension(s) active)"
+    )
+    sorted
   }
 
+  private def elapsedMs(t0: Long): Long = (System.nanoTime() - t0) / 1000000L
+
   /** Find extension that can handle a statement.
     */
   def findHandler(statement: Statement): Option[ExtensionSpi] = {

core/src/test/scala/app/softnetwork/elastic/client/ExtensionEagerInitSpec.scala

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2025 SOFTNETWORK
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package app.softnetwork.elastic.client
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.slf4j.{Logger, LoggerFactory}
+
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext, Future}
+
+/** #163 — eager extension init. Two guarantees:
+  *   1. `initializeExtensions()` forces the ServiceLoader scan and reports the count (core's own
+  *      META-INF/services registers CoreDdlExtension + CoreDqlExtension → always >= 2 here).
+  *   1. THE TRAP — a delegator forwards to its delegate — the wrapper's own (inherited, unused)
+  *      registry must NOT be the one warmed, because run() delegates to delegate.run().
+  */
+class ExtensionEagerInitSpec extends AnyFlatSpec with Matchers {
+
+  implicit val ec: ExecutionContext = ExecutionContext.global
+
+  private val testLogger: Logger = LoggerFactory.getLogger(getClass)
+
+  // `protected def logger` is the ONLY abstract member NopeClientApi leaves open; `config` and
+  // `metrics` have concrete defaults, so licenseRefreshStrategy resolves from ConfigFactory.load()
+  // → Community fallback, and clusterUuid/clusterName/version no-op failures are absorbed by
+  // ExtensionApi's `case _ =>`. Same recipe as CoreDqlExtensionSpec's RecordingClient.
+  private def newNopeClient(): ElasticClientApi = new NopeClientApi {
+    override protected def logger: Logger = testLogger
+  }
+
+  "initializeExtensions" should "force the registry and report the loaded-extension count" in {
+    val client = newNopeClient()
+    val n = Await.result(client.initializeExtensions(), 30.seconds)
+    n should be >= 2 // CoreDdlExtension + CoreDqlExtension from core's META-INF/services
+  }
+
+  it should "be forwarded to the delegate by ElasticClientDelegator" in {
+    @volatile var forwarded = false
+    val probe: ElasticClientApi = new NopeClientApi {
+      override protected def logger: Logger = testLogger
+      override def initializeExtensions()(implicit ec: ExecutionContext): Future[Int] = {
+        forwarded = true
+        Future.successful(42)
+      }
+    }
+    val wrapper = new ElasticClientDelegator {
+      override val delegate: ElasticClientApi = probe
+    }
+    Await.result(wrapper.initializeExtensions(), 5.seconds) shouldBe 42
+    forwarded shouldBe true
+  }
+}