feature: Add portable idle/read timeout to the Native HTTP server

Replace the dropped SO_RCVTIMEO (macOS timeval layout bug) with poll()-based
timeouts (posixlib's poll uses a millisecond CInt — no timeval trap, portable).
Fixes the long-standing limitation where an idle or silently-dropped client
pinned a worker thread forever.

- NativeSocket: waitReadable(fd, timeoutMillis) via poll (1 readable / 0 timeout
  / -1 hangup, checking POLLHUP/POLLERR/POLLNVAL); enableKeepAlive (SO_KEEPALIVE).
- NativeServerConfig: idleTimeoutMillis + readTimeoutMillis (default 30s) + builders.
- HttpConnectionReader: readChunk thunk takes an `idle` flag (empty buffer =
  awaiting a new request) so the server applies the idle vs read timeout.
- handleConnection: enableKeepAlive + polling read thunk; an idle keep-alive
  connection closes after idleTimeoutMillis, a stalled mid-request after
  readTimeoutMillis.
- streamSse: probe the socket between latch waits so an idle client disconnect is
  detected and the subscription cancelled (frees the worker), instead of parking
  until the next failed write.

WebSocket clean disconnects already unblock its blocking recv; half-open WS
heartbeat stays a follow-up (SO_KEEPALIVE helps at the OS level).

Tests: idle keep-alive closed after timeout; incomplete request -> 400 after
read timeout. uniNative/test: 1399 tests, 0 failed.

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

Author: xerial

plans/2026-06-19-native-idle-timeout.md

@@ -0,0 +1,35 @@
+# Native HTTP server — portable idle/read timeout
+
+## Context
+The Native server (#574) used blocking `recv` with no timeout — `SO_RCVTIMEO` was dropped because
+macOS's `timeval`/`suseconds_t` layout differs from posixlib's, making `recv` return immediately. So
+an idle or silently-dropped client pinned a worker thread indefinitely (the recurring SSE/keep-alive
+limitation noted across #574/#575/#576).
+
+Fix: use `poll()` (posixlib 0.5.12), whose timeout is a plain millisecond `CInt` — no `timeval` trap,
+portable across macOS and Linux.
+
+## Changes (all Native)
+- **`NativeSocket`** — `waitReadable(fd, timeoutMillis): Int` (1 readable / 0 timeout / -1 hangup,
+  checking `POLLHUP|POLLERR|POLLNVAL`); `enableKeepAlive(fd)` (`SO_KEEPALIVE`, best-effort OS reaping
+  of dead peers).
+- **`NativeServerConfig`** — `idleTimeoutMillis` (default 30000) + `readTimeoutMillis` (30000) +
+  `withIdleTimeoutMillis`/`withReadTimeoutMillis`.
+- **`HttpConnectionReader`** — the `readChunk` thunk now takes an `idle` flag (true when the buffer is
+  empty = waiting for a new request), so the caller applies the idle vs read timeout.
+- **`NativeHttpServer.handleConnection`** — `enableKeepAlive`, then a polling read thunk: idle
+  keep-alive waits `idleTimeoutMillis`, mid-request reads wait `readTimeoutMillis`; a timeout/hangup
+  yields an empty chunk → the reader ends the connection (idle → Closed, mid-request → 400).
+- **`streamSse`** — replaces the bare `done.await()` with `done.await(idleTimeoutMillis)` intervals,
+  probing the socket (non-blocking `waitReadable`) between waits; a disconnected peer cancels the
+  subscription so the worker is freed instead of parked until the next failed write.
+
+WebSocket is unchanged: a clean disconnect already unblocks its blocking `recv` (returns 0). Half-open
+WS detection (ping/pong) remains a separate follow-up; `SO_KEEPALIVE` helps at the OS level.
+
+## Verification
+`uniNative/test` (1399 tests, 0 failed). New tests: an idle keep-alive connection is closed after
+`idleTimeoutMillis`; an incomplete request is closed (400) after `readTimeoutMillis`.
+
+## Follow-ups (remaining)
+Native libcurl client bug; cross-platform WebSocket client; WS ping/pong heartbeat; permessage-deflate.

uni/.native/src/main/scala/wvlet/uni/http/NativeHttpProtocol.scala

@@ -30,17 +30,19 @@ private[http] enum ReadResult:
   * by `Content-Length`; chunked request bodies are not supported (MVP) and yield a `BadRequest`.
   *
   * @param readChunk
-  *   reads the next batch of bytes from the socket; an empty array means EOF.
+  *   reads the next batch of bytes from the socket; an empty array means EOF or timeout. The `idle`
+  *   argument is true when waiting for the first bytes of a new request (the buffer is empty) and
+  *   false once a request has started arriving, so the caller can apply an idle vs a read timeout.
   */
-private[http] class HttpConnectionReader(readChunk: () => Array[Byte], maxRequestSize: Int):
+private[http] class HttpConnectionReader(readChunk: Boolean => Array[Byte], maxRequestSize: Int):
 
   // Amortized-O(1) append buffer (indexed scan, single slice per request) — avoids the O(n^2) of
   // repeatedly concatenating Array[Byte].
   private val buf = mutable.ArrayBuffer.empty[Byte]
 
-  /** Append another chunk to the buffer. Returns false on EOF. */
+  /** Append another chunk to the buffer. Returns false on EOF/timeout. */
   private def fill(): Boolean =
-    val chunk = readChunk()
+    val chunk = readChunk(buf.isEmpty)
     if chunk.isEmpty then
       false
     else

uni/.native/src/main/scala/wvlet/uni/http/NativeHttpServer.scala

@@ -95,8 +95,23 @@ class NativeHttpServer(config: NativeServerConfig) extends HttpServer with LogSu
 
   private def handleConnection(clientFd: Int): Unit =
     try
+      NativeSocket.enableKeepAlive(clientFd)
+      // Poll before recv so an idle/slow/disconnected client can't pin a worker forever: an idle
+      // keep-alive connection waits up to idleTimeoutMillis for the next request; once a request has
+      // started arriving, subsequent reads wait up to readTimeoutMillis. A timeout (or hangup) maps
+      // to an empty chunk, which the reader treats as end-of-connection.
       val reader = HttpConnectionReader(
-        () => NativeSocket.recvChunk(clientFd),
+        idle =>
+          val timeout =
+            if idle then
+              config.idleTimeoutMillis
+            else
+              config.readTimeoutMillis
+          if NativeSocket.waitReadable(clientFd, timeout) == 1 then
+            NativeSocket.recvChunk(clientFd)
+          else
+            Array.emptyByteArray
+        ,
         config.maxRequestSize
       )
       var continue = true
@@ -164,10 +179,9 @@ class NativeHttpServer(config: NativeServerConfig) extends HttpServer with LogSu
     * long-lived by design, so no handler-await timeout is applied here). A failed chunk write
     * cancels the subscription. Returns whether the connection is still usable for keep-alive.
     *
-    * Limitation (blocking model): a client that disconnects while the stream is *idle* (no event to
-    * write) is not detected until the next event fails to write, so the worker stays parked in
-    * `done.await()` — each live stream pins one worker thread. A portable idle/read timeout is a
-    * follow-up (see the SSE-streaming plan).
+    * The worker is parked until the stream terminates, but it periodically probes the socket (every
+    * idleTimeoutMillis) so a client that disconnects while the stream is *idle* is detected and the
+    * subscription cancelled, rather than leaking the worker until the next event fails to write.
     */
   private def streamSse(clientFd: Int, response: Response, keepAlive: Boolean): Boolean =
     if !NativeSocket.sendAll(clientFd, HttpResponseWriter.serializeSseHeaders(response, keepAlive))
@@ -205,9 +219,23 @@ class NativeHttpServer(config: NativeServerConfig) extends HttpServer with LogSu
             done.countDown()
         }
       )
-      done.await()
+      // Wait for the stream to terminate, but probe the socket between waits so an idle client
+      // disconnect is detected promptly (a disconnected peer reads as readable-then-EOF or a hangup).
+      var alive = true
+      while alive && !done.await(config.idleTimeoutMillis.toLong, TimeUnit.MILLISECONDS) do
+        NativeSocket.waitReadable(clientFd, 0) match
+          case -1 =>
+            alive = false // peer hung up
+          case 1 =>
+            // SSE clients shouldn't send; a readable socket that yields no bytes is EOF (disconnect).
+            if NativeSocket.recvChunk(clientFd).isEmpty then
+              alive = false
+          case _ =>
+            () // genuinely idle but still connected
+      if !alive then
+        subscription.cancel
       // Terminate the chunked body if the client is still there.
-      ok.get() && NativeSocket.sendAll(clientFd, HttpResponseWriter.finalChunk)
+      ok.get() && alive && NativeSocket.sendAll(clientFd, HttpResponseWriter.finalChunk)
 
   /**
     * Run the handler and block for its single response. The handler may be asynchronous (Rx), so

uni/.native/src/main/scala/wvlet/uni/http/NativeServer.scala

@@ -58,6 +58,10 @@ case class NativeServerConfig(
     workerThreads: Int = 16,
     // Max time to wait for a handler's Rx to produce a response before returning 503
     handlerTimeoutMillis: Long = 30000,
+    // Max time a kept-alive connection may sit idle (awaiting the next request) before it is closed
+    idleTimeoutMillis: Int = 30000,
+    // Max time to wait for more bytes once a request has started arriving before closing
+    readTimeoutMillis: Int = 30000,
     // WebSocket routes, matched by path during the HTTP upgrade handshake
     override val webSocketRoutes: Seq[WebSocketRoute] = Nil,
     // Maximum size (bytes) of an inbound WebSocket message
@@ -104,6 +108,14 @@ case class NativeServerConfig(
     require(millis > 0, "handlerTimeoutMillis must be positive")
     copy(handlerTimeoutMillis = millis)
 
+  def withIdleTimeoutMillis(millis: Int): NativeServerConfig =
+    require(millis > 0, "idleTimeoutMillis must be positive")
+    copy(idleTimeoutMillis = millis)
+
+  def withReadTimeoutMillis(millis: Int): NativeServerConfig =
+    require(millis > 0, "readTimeoutMillis must be positive")
+    copy(readTimeoutMillis = millis)
+
   /** Register a WebSocket route; the factory creates a fresh handler per accepted connection. */
   def withWebSocketRoute(path: String)(
       handlerFactory: Request => WebSocketHandler

uni/.native/src/main/scala/wvlet/uni/http/NativeSocket.scala

@@ -17,6 +17,8 @@ import scala.scalanative.libc.string as cstring
 import scala.scalanative.posix.arpa.inet
 import scala.scalanative.posix.netinet.in.{in_addr, sockaddr_in}
 import scala.scalanative.posix.netinet.inOps.*
+import scala.scalanative.posix.poll.*
+import scala.scalanative.posix.pollOps.*
 import scala.scalanative.posix.sys.socket as csocket
 import scala.scalanative.posix.sys.socket.{sockaddr, socklen_t}
 import scala.scalanative.posix.unistd
@@ -116,6 +118,42 @@ private[http] object NativeSocket:
     null.asInstanceOf[Ptr[socklen_t]]
   )
 
+  /** Enable TCP keep-alive so the OS eventually reaps a silently-dropped peer (best effort). */
+  def enableKeepAlive(fd: Int): Unit =
+    val optval = stackalloc[CInt]()
+    !optval = 1
+    csocket.setsockopt(
+      fd,
+      csocket.SOL_SOCKET,
+      csocket.SO_KEEPALIVE,
+      optval.asInstanceOf[Ptr[Byte]],
+      sizeof[CInt].toUInt
+    )
+
+  /**
+    * Wait until `fd` is readable or `timeoutMillis` elapses, using `poll` (a portable millisecond
+    * timeout — avoids the macOS `SO_RCVTIMEO`/`timeval` layout trap). Returns 1 if readable, 0 on
+    * timeout, -1 if the peer hung up or `poll` errored.
+    */
+  def waitReadable(fd: Int, timeoutMillis: Int): Int =
+    val fds = stackalloc[struct_pollfd]()
+    fds.fd = fd
+    fds.events = POLLIN.toShort
+    fds.revents = 0.toShort
+    val rc = poll(fds, 1.toUInt, timeoutMillis)
+    if rc < 0 then
+      -1
+    else if rc == 0 then
+      0
+    else
+      val re = fds.revents.toInt
+      if (re & (POLLHUP | POLLERR | POLLNVAL)) != 0 then
+        -1
+      else if (re & POLLIN) != 0 then
+        1
+      else
+        0
+
   /**
     * Receive up to ChunkSize bytes. Returns an empty array on EOF or error (caller treats both as
     * end-of-connection).

uni/.native/src/test/scala/wvlet/uni/http/NativeServerTest.scala

@@ -292,6 +292,44 @@ class NativeServerTest extends UniTest:
       }
   }
 
+  test("should close an idle keep-alive connection after the idle timeout") {
+    NativeServer
+      .withHandler(_ => Response.ok("ok"))
+      .withIdleTimeoutMillis(300)
+      .withPort(0)
+      .start { server =>
+        val fd = connectLoopback(server.localPort)
+        try
+          // Keep-alive request (no Connection: close), so the server waits for a follow-up request.
+          NativeSocket.sendAll(
+            fd,
+            "GET /x HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1)
+          )
+          NativeSocket.recvChunk(fd).nonEmpty shouldBe true // the 200 response
+          // Now stay idle; the server should close the connection after the idle timeout.
+          val start = System.currentTimeMillis()
+          var chunk = NativeSocket.recvChunk(fd)
+          while chunk.nonEmpty do
+            chunk = NativeSocket.recvChunk(fd)
+          val elapsed = System.currentTimeMillis() - start
+          (elapsed < 5000) shouldBe true
+        finally
+          NativeSocket.close(fd)
+      }
+  }
+
+  test("should close a connection with an incomplete request after the read timeout") {
+    NativeServer
+      .withHandler(_ => Response.ok("ok"))
+      .withReadTimeoutMillis(300)
+      .withPort(0)
+      .start { server =>
+        // Headers are never terminated, so the read times out mid-request.
+        val response = request(server.localPort, "GET /x HTTP/1.1\r\nHost: localhost\r\n")
+        response.status shouldBe 400
+      }
+  }
+
   // ---- WebSocket ----
 
   /**