diff --git a/src/main/scala/ccas/server/CcasServer.scala b/src/main/scala/ccas/server/CcasServer.scala index 6c459e1..7b28b4f 100644 --- a/src/main/scala/ccas/server/CcasServer.scala +++ b/src/main/scala/ccas/server/CcasServer.scala @@ -58,9 +58,14 @@ object CcasServer extends ZIOAppDefault { JobRunner.live, JobScheduler.live, // Read-idle reaper for keep-alive connections (zio-http's `Server.Config.default` leaves `idleTimeout=None`), - // so a client that vanishes without FIN/RST can't pin a Netty channel + fd forever. Set above - // `JobLogStream.KeepAliveInterval` (20s) so a streaming `/api/jobs/{id}/logs` follow's keepalive resets the - // timer before it fires; only genuinely dead connections get reaped. + // so a client that vanishes without FIN/RST can't pin a Netty channel + fd forever. This installs a *read-only* + // Netty `ReadTimeoutHandler` (`ServerChannelInitializer`): it resets on inbound (client→server) reads only, never + // on writes. A streaming `/api/jobs/{id}/{logs,progress}` follow is write-only server→client once the GET is sent, + // so its `JobLogStream` keepalive ticks (outbound) can't reset this timer — every live follow is reaped on the 60s + // schedule and the follower transparently reconnects (#161). That reap's `ReadTimeoutException` surfaces as a + // "Fatal exception in Netty" WARN which is benign noise, filtered at the logger (`ProgressDisplay.isBenignReadIdleReap`); + // this reaper's real remaining job is bounding idle *non-streaming* keep-alive fds. zio-http exposes no per-route + // idleTimeout nor a server `SO_KEEPALIVE` option, so one global read-idle timeout is the only knob available. Server.defaultWith(_.binding(host, port).idleTimeout(60.seconds)) ) } diff --git a/src/main/scala/ccas/utils/ProgressDisplay.scala b/src/main/scala/ccas/utils/ProgressDisplay.scala index 803014a..67e7483 100644 --- a/src/main/scala/ccas/utils/ProgressDisplay.scala +++ b/src/main/scala/ccas/utils/ProgressDisplay.scala @@ -7,6 +7,7 @@ import java.time.{Instant, ZoneId} import java.util.concurrent.atomic.AtomicInteger import java.util.regex.Pattern +import io.netty.handler.timeout.ReadTimeoutException import zio.stream.SubscriptionRef import zio.{Cause, FiberId, FiberRef, FiberRefs, LogLevel, LogSpan, Ref, Runtime, Scope, Trace, UIO, Unsafe, URIO, URLayer, ZIO, ZLayer, ZLogger} @@ -204,8 +205,13 @@ final class ProgressDisplay private[utils] ( val threshold = context.getOrDefault(FiberRef.currentLogLevel) if (level >= threshold) { try { - val sink = context.getOrDefault(JobLogSink.currentSink) - logAboveBarsSync(sink, ProgressDisplay.format(level, message(), Instant.now(), cause, spans, annotations)) + // Force the message thunk once, inside the try so a throwing thunk stays caught (see `testLoggerSwallowsThrow`), + // and reuse it for both the read-idle-reap filter and the formatter. + val msg = message() + if (!ProgressDisplay.isBenignReadIdleReap(level, msg, cause)) { + val sink = context.getOrDefault(JobLogSink.currentSink) + logAboveBarsSync(sink, ProgressDisplay.format(level, msg, Instant.now(), cause, spans, annotations)) + } } catch { case t: Throwable => t.printStackTrace(err) } } } @@ -460,6 +466,30 @@ object ProgressDisplay { private def formatTime(when: Instant): String = when.atZone(Zone).toLocalTime.format(TimeFormat) + /** zio-http's `ServerInboundHandler.exceptionCaught` logs every fatal channel exception under this exact message. */ + private val FatalNettyExceptionMessage = "Fatal exception in Netty" + + /** True for the benign WARN zio-http emits when the server's read-idle reaper closes a live streaming follow. + * + * `CcasServer` sets `Server.Config#idleTimeout`, which installs a read-only Netty `ReadTimeoutHandler`. A job-log / + * progress follow (`GET /api/jobs/{id}/{logs,progress}`) is write-only server→client once the request is sent, so the + * read timer is never reset and the handler reaps every live follow on schedule — surfacing as zio-http's + * `ZIO.logWarningCause("Fatal exception in Netty", Cause.die(ReadTimeoutException))`. The follow's client transparently + * reconnects (#161), so the reap is benign transport noise, not a failure, and we drop exactly this WARN. + * + * The match is deliberately over-specified — the exact zio-http message AND a cause that is *solely* a + * `ReadTimeoutException` defect (no typed failure, no other defect). This scopes it to the server reaper rather than + * any bare read-timeout WARN elsewhere, and makes both coordinates fail *open*: if a zio-http upgrade renames the + * message or wraps the exception, the filter simply stops matching and the (still benign) WARN reappears — noise + * returns, no genuine signal is ever hidden. + */ + private def isBenignReadIdleReap(level: LogLevel, message: String, cause: Cause[Any]): Boolean = + level == LogLevel.Warning && + message == FatalNettyExceptionMessage && + cause.failures.isEmpty && + cause.defects.nonEmpty && + cause.defects.forall(_.isInstanceOf[ReadTimeoutException]) + private def format( level: LogLevel, msg: String, diff --git a/src/test/scala/ccas/utils/TestProgressBar.scala b/src/test/scala/ccas/utils/TestProgressBar.scala index da2edf9..690254e 100644 --- a/src/test/scala/ccas/utils/TestProgressBar.scala +++ b/src/test/scala/ccas/utils/TestProgressBar.scala @@ -2,8 +2,9 @@ package ccas.utils import java.io.{ByteArrayOutputStream, PrintStream} +import io.netty.handler.timeout.ReadTimeoutException import zio.stream.SubscriptionRef -import zio.{LogLevel, Promise, Ref, ZIO} +import zio.{Cause, LogLevel, Promise, Ref, ZIO} import zio.test.{assertCompletes, assertTrue, Spec, ZIOSpecDefault} object TestProgressBar extends ZIOSpecDefault { @@ -25,7 +26,8 @@ object TestProgressBar extends ZIOSpecDefault { testCurrentLogLevelEnablesDebug, testLoggerSwallowsThrow, testSourcedRendersPrefixTag, - testStripSourceTag + testStripSourceTag, + testBenignReadIdleReapFiltered ) // No `@@ TestAspect.sequential`: each test owns its capture buffers and installs loggers/sinks via fiber-scoped // FiberRefs (`currentLoggers`, `currentSink`), so the suite holds no process-global state to serialise. (Cross-suite @@ -356,6 +358,31 @@ object TestProgressBar extends ZIOSpecDefault { ) } + /** The server's read-idle reaper closes every live streaming follow on schedule (a read-only `ReadTimeoutHandler` + * can't be reset by the outbound keepalive), which zio-http logs as a benign `Cause.die(ReadTimeoutException)` WARN. + * The logger drops exactly that WARN (`ProgressDisplay.isBenignReadIdleReap`), but any other fatal-Netty WARN — a + * defect of a different type — must still be rendered. Both are emitted here; only the second should survive. + */ + private def testBenignReadIdleReapFiltered = + test("read-idle-reap WARN is filtered while other fatal-Netty WARNs still log") { + val effect = + ZIO.logWarningCause("Fatal exception in Netty", Cause.die(ReadTimeoutException.INSTANCE)) *> + ZIO.logWarningCause("genuine netty fault", Cause.die(new RuntimeException("kaboom"))) *> + ZIO.logWarningCause("read timeout we DO want", Cause.die(ReadTimeoutException.INSTANCE)) + withLogCapture(effect).map { case (_, out) => + val clean = stripAnsi(out) + assertTrue( + // The exact zio-http server-reaper WARN (its message + a bare ReadTimeoutException) is dropped... + !clean.contains("Fatal exception in Netty"), + // ...but a different fatal-Netty WARN survives... + clean.contains("genuine netty fault"), + clean.contains("kaboom"), + // ...and so does a bare ReadTimeoutException under a different message — the filter is scoped, not blanket. + clean.contains("read timeout we DO want") + ) + } + } + private def testLoggerSwallowsThrow = test("asLogger swallows exceptions from a throwing message thunk") { ZIO.suspendSucceed { val outBaos = new ByteArrayOutputStream