Annotate slow paths, fix multithreading paths

.github/workflows/build.yml

@@ -7,6 +7,10 @@ jobs:
     runs-on: ubuntu-24.04
     steps:
       - uses: actions/checkout@v5
+      - uses: actions/setup-java@v3
+        with:
+          distribution: "temurin"
+          java-version: "17"
       - uses: coursier/cache-action@v7
       - name: Install Dependencies
         run: |
@@ -25,6 +29,8 @@ jobs:
         run: ./mill __.compile
       - name: Publish libraries locally
         run: ./mill __.publishLocal
+      - name: Link all test binaries
+        run: ./mill integration.tests.__.nativeLink
       - name: Run Unit Tests
         run: ./mill snunit.test
       - name: Run Integration Tests
@@ -36,6 +42,7 @@ jobs:
       - uses: actions/checkout@v5
         with:
           fetch-depth: 0
+      - uses: coursier/cache-action@v8
       - name: Check Binary Compatibility
         run: ./mill __.mimaReportBinaryIssues
 
@@ -53,7 +60,11 @@ jobs:
       LC_ALL: "en_US.UTF-8"
     steps:
       - uses: actions/checkout@v5
-      - uses: coursier/cache-action@v7
+      - uses: actions/setup-java@v3
+        with:
+          distribution: "temurin"
+          java-version: "17"
+      - uses: coursier/cache-action@v8
       - name: Publish to Maven Central
         run: |
           if [[ $(git tag --points-at HEAD) != '' ]]; then

build.mill

@@ -1,4 +1,5 @@
-//| mill-version: 1.1.1
+//| mill-version: 1.1.2
+//| mill-jvm-version: system
 //| mvnDeps:
 //|   - com.goyeau::mill-scalafix::0.6.0
 //|   - com.lihaoyi::mill-contrib-buildinfo:$MILL_VERSION

integration/test/src/BaseTests.scala

@@ -1,43 +1,68 @@
 package snunit.test
 
 import utest._
+import scala.concurrent.duration._
 
 object BaseTests extends TestSuite {
   val tests = Tests {
+
     test("hello-world") {
-      withDeployedExample("hello-world") {
-        locally {
+      val helloWorldExample = Example("hello-world")
+      test("hello") {
+        helloWorldExample.running {
           val result = request.get(baseUrl).text()
           val expectedResult = "Hello world!\n"
           assert(result == expectedResult)
         }
-        locally {
+      }
+      test("version") {
+        helloWorldExample.running {
           val result = request.get(uri"$baseUrl/version").text()
           val expectedResult = "HTTP/1.1"
           assert(result == expectedResult)
         }
-        locally {
+      }
+      test("target") {
+        helloWorldExample.running {
           val result = request.get(baseUrl.withPath("target", "%2F%2f%5C%5c").pathSegmentsEncoding(identity)).text()
           val expectedResult = "/target/%2F%2f%5C%5c"
           assert(result == expectedResult)
         }
-        locally {
+      }
+      test("path") {
+        helloWorldExample.running {
           val result =
             request.get(baseUrl.withPath("path", "foo%2Fbar%2f%5C%5c").pathSegmentsEncoding(identity)).text()
           val expectedResult = """/path/foo/bar/\\"""
           assert(result == expectedResult)
         }
-        locally {
+      }
+      test("empty") {
+        helloWorldExample.running {
           val result = request.get(uri"$baseUrl/empty").text()
           val expectedResult = ""
           assert(result == expectedResult)
         }
-        locally {
+      }
+      test("async") {
+        helloWorldExample.running {
+          /* Hit /async multiple times on same client to exercise keep-alive and handler_done pipe (no double wake). */
+          val expectedResult = "Hello world!\n"
+          (1 to 50).foreach { _ =>
+            val result = request.get(uri"$baseUrl/async").text()
+            assert(result == expectedResult)
+          }
+        }
+      }
+      test("echo") {
+        helloWorldExample.running {
           val result = request.post(uri"$baseUrl/echo").body("hello").text()
           val expectedResult = "hello"
           assert(result == expectedResult)
         }
-        locally {
+      }
+      test("headers") {
+        helloWorldExample.running {
           val responseHeaders = request
             .get(uri"$baseUrl/headers")
             .header("foo", "bar")
@@ -51,6 +76,62 @@ object BaseTests extends TestSuite {
           assert(responseHeaders.contains(Header("bla", "bal")))
         }
       }
+      test("close") {
+        helloWorldExample.running {
+          /* Reproduce wrk-style termination: connect, send GET /async, close socket without reading.
+           * Server sees client disconnect (FIN/RST) while async handler may still be running.
+           * Use many concurrent connections (like wrk -d 1) so the server gets a burst of closes. */
+          val host = baseUrl.host.get
+          val port = baseUrl.port.get
+          val requestBytes =
+            s"GET /async HTTP/1.1\r\nHost: $host:$port\r\nConnection: keep-alive\r\n\r\n".getBytes(
+              java.nio.charset.StandardCharsets.US_ASCII
+            )
+
+          def connectSendAndClose(): Unit = {
+            val socket = new java.net.Socket()
+            try {
+              socket.connect(new java.net.InetSocketAddress(host, port), 5000)
+              socket.setSoTimeout(1000)
+              socket.getOutputStream.write(requestBytes)
+              socket.getOutputStream.flush()
+              // Close immediately without reading response (like wrk on process exit).
+              socket.close()
+            } catch { case _: Exception => /* ignore */ }
+            finally
+              if (!socket.isClosed)
+                try socket.close()
+                catch { case _: Exception => }
+          }
+
+          // Many concurrent connections that all close without reading (like wrk exiting after -d 1).
+          val n = 15
+          val threads = (1 to n).map(_ =>
+            new Thread(() => connectSendAndClose(), "close-test-client")
+          )
+          threads.foreach(_.start())
+          threads.foreach(_.join())
+
+          // Give async handlers time to complete and server to drain handler_pipe.
+          Thread.sleep(500)
+
+          // Server must still be alive: a normal request must succeed.
+          val result = request.get(uri"$baseUrl/async").readTimeout(5.seconds).text()
+          assert(result == "Hello world!\n")
+        }
+      }
+      test("closeAfterWrk") {
+        /* Reproduce exact user scenario: run real wrk -d 1, then server must still answer.
+         * This test fails when the server gets stuck after wrk exits (closes all connections).
+         * Requires: port 8080 free (kill any stuck server first), wrk installed. */
+        helloWorldExample.running {
+          /* Run wrk; ignore exit code (e.g. connection refused if server slow to start). */
+          os.proc("wrk", "-d", "1", "-t", "2", "-c", "10", s"http://localhost:8080/async").call(check = false)
+          Thread.sleep(3000) /* let async handlers complete and server drain handler_pipe */
+          val result = request.get(uri"$baseUrl/async").readTimeout(10.seconds).text()
+          assert(result == "Hello world!\n")
+        }
+      }
     }
     test("multiple-handlers") {
       withDeployedExample("multiple-handlers") {

integration/test/src/utils.scala

@@ -10,7 +10,7 @@ import sttp.client3.HttpClientFutureBackend
 private def runMillCommand(command: String) = os
   .proc(
     "./mill",
-    // adding `-i` breaks the ability to close unitd processes
+    "-i",
     "--no-build-lock",
     "--ticker",
     "false",
@@ -20,15 +20,21 @@ private def runMillCommand(command: String) = os
   .call(
     cwd = os.Path(sys.env("MILL_WORKSPACE_ROOT"))
   )
+class Example(projectName: String, crossSuffix: String = "") {
+  private val Vector(s"\"$_:$_:$_:$nativeBinary\"") =
+    runMillCommand(s"integration.tests.$projectName$crossSuffix.nativeLink").out.lines(): @unchecked
+  private val workspace = os.Path(sys.env("MILL_WORKSPACE_ROOT"))
+
+  def running[T](f: => T): T = {
+    val process2 = os.proc(nativeBinary).spawn(cwd = workspace)
+    Thread.sleep(1000)
+    try { f }
+    finally { process2.close() }
+  }
+}
 
 def withDeployedExample[T](projectName: String, crossSuffix: String = "")(f: => T): T = {
-  val Vector(s"\"$_:$_:$_:$nativeBinary\"") =
-    runMillCommand(s"integration.tests.$projectName$crossSuffix.nativeLink").out.lines(): @unchecked
-  val workspace = os.Path(sys.env("MILL_WORKSPACE_ROOT"))
-  val process2 = os.proc(nativeBinary).spawn(cwd = workspace)
-  Thread.sleep(1000)
-  try { f }
-  finally { process2.close() }
+  Example(projectName, crossSuffix).running(f)
 }
 def withDeployedExampleHttp4s(projectName: String)(f: => Unit) = {
   BuildInfo.http4sVersions.split(':').foreach { versions =>

integration/tests/hello-world/src/snunit/tests/HelloWorld.scala

@@ -18,6 +18,16 @@ object MyHandler extends RequestHandler {
           content = "Request headers",
           headers = req.headers
         )
+
+      case Method.GET -> "/async" =>
+        concurrent.ExecutionContext.global.execute(() => {
+          req.send(
+            statusCode = StatusCode.OK,
+            content = "Hello world!\n",
+            headers = Headers("Content-Type" -> "text/plain")
+          )
+        })
+
       case Method.GET -> path =>
         val content =
           if (path.startsWith("/path")) req.path

snunit-undertow/src/io/undertow/server/util/HeaderValues.scala

@@ -7,8 +7,16 @@ final class HeaderValues private[undertow] (key: String, value: String)
     with java.util.Deque[String]
     with java.util.List[String] {
   def getHeaderName(): String = key
-  def descendingIterator(): java.util.Iterator[String] = ???
-  def element(): String = ???
+  def descendingIterator(): java.util.Iterator[String] = Array(value).iterator.asJava
+  def element(): String = value
+
+  // Deque methods (single-element: get/peek return value; add/remove throw)
+  def addFirst(x$0: String): Unit = throw new UnsupportedOperationException
+  def addLast(x$0: String): Unit = throw new UnsupportedOperationException
+  def getFirst(): String = value
+  def getLast(): String = value
+  def removeFirst(): String = throw new UnsupportedOperationException
+  def removeLast(): String = throw new UnsupportedOperationException
 
   // Members declared in java.util.List
   def add(x$1: Int, x$2: String): Unit = ???
@@ -37,5 +45,5 @@ final class HeaderValues private[undertow] (key: String, value: String)
   def remove(): String = ???
   def removeFirstOccurrence(x$0: Object): Boolean = ???
   def removeLastOccurrence(x$0: Object): Boolean = ???
-  override def reversed(): HeaderValues = ???
+  def reversed(): HeaderValues = this
 }

snunit/resources/scala-native/snunit/README.md

@@ -9,6 +9,14 @@ This directory contains a **minimal in-process** implementation of the NGINX Uni
   - `nxt_unit_typedefs.h`, `nxt_unit_sptr.h`, `nxt_unit_field.h`, `nxt_unit_request.h`, `nxt_unit_response.h`, `nxt_unit.h`
 - **nxt_auto_config.h**, **nxt_version.h** – Minimal stubs for the embed build (no Unit `configure`).
 
+## Vendoring from NGINX Unit
+
+To refresh the Unit API headers from a local Unit tree (e.g. `/Users/lorenzo/scala/unit`), copy from `unit/src/` into this directory:
+
+- `nxt_unit.h`, `nxt_unit_typedefs.h`, `nxt_unit_request.h`, `nxt_unit_response.h`, `nxt_unit_field.h`, `nxt_unit_sptr.h`, `nxt_unit_websocket.h`, `nxt_websocket_header.h`
+
+Do **not** overwrite `nxt_auto_config.h` or `nxt_version.h` (snunit keeps minimal stubs). The embed uses `nxt_unit_init(init, host, port)` (3 args); Unit’s public API uses 1 arg and reads from `NXT_UNIT_INIT` env, so `nxt_unit.h` is adjusted for the embed. `NXT_UNIT_HASH_HOST` in `nxt_unit_field.h` is added for the embed’s Host-header parsing.
+
 ## Build
 
 Scala Native compiles all `.c` (and `.cpp`) files under `src/main/resources/scala-native` (or `resources/scala-native` for the snunit library) and links them into the final binary. No extra build step is needed.

snunit/resources/scala-native/snunit/nxt_auto_config.h

@@ -21,4 +21,24 @@
 
 #define NXT_DEBUG 0
 
+/* Endianness for nxt_websocket_header.h (from Unit). */
+#if defined(__BYTE_ORDER__) && __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__
+#define NXT_HAVE_BIG_ENDIAN 1
+#define NXT_HAVE_LITTLE_ENDIAN 0
+#else
+#define NXT_HAVE_BIG_ENDIAN 0
+#define NXT_HAVE_LITTLE_ENDIAN 1
+#endif
+
+/* Branch prediction hints (from unit src/nxt_clang.h). */
+#if defined(__GNUC__) || defined(__clang__)
+#define nxt_expect(c, x)   __builtin_expect((long) (x), (c))
+#define nxt_fast_path(x)   nxt_expect(1, x)
+#define nxt_slow_path(x)   nxt_expect(0, x)
+#else
+#define nxt_expect(c, x)   (x)
+#define nxt_fast_path(x)   (x)
+#define nxt_slow_path(x)   (x)
+#endif
+
 #endif