He-Pin · Nov 1, 2026 · Oct 30, 2026
diff --git a/docs/src/main/paradox/release-notes/releases-2.0.md b/docs/src/main/paradox/release-notes/releases-2.0.md
@@ -48,6 +48,16 @@ Please read the @ref:[Migration Guide](../migration/migration-guide-1.x-2.x.md)
 breaking API changes, dependency changes (such as the Agrona 2.x `--add-opens` requirement) and the configuration
 defaults that have changed in Pekko 2.0.0.
 
+Unless already waiting for the peer's `close_notify`, cancelling a TLS operator's transport output
+now reports `SessionTruncated` on its plaintext output
+if a session was established and the peer's `close_notify` has not been processed, including when
+no inbound data is buffered or a renegotiation is in progress ([PR3740](https://github.com/apache/pekko/pull/3740)).
+Previously this path could complete silently after discarding received records. Consumers using
+the TLS `BidiFlow` directly must handle `SessionTruncated` as well as `SessionBytes`. The Scala and
+Java `Tcp.bindWithTls` and `Tcp.outgoingConnectionWithTls` APIs filter out this signal and continue
+to expose `ByteString` streams. The final signal obeys backpressure; cancel the plaintext output
+too if it will no longer be consumed. See @ref:[Stream IO](../stream/stream-io.md) for details.
+
 ### Bug Fixes
 
 * Replaying a replicated event sourced actor from a snapshot caused a bug ([PR2372](https://github.com/apache/pekko/pull/2372))

diff --git a/docs/src/main/paradox/stream/stream-io.md b/docs/src/main/paradox/stream/stream-io.md
@@ -203,6 +203,39 @@ The drain also continues after a TLS 1.3 peer sends `close_notify`, because that
 peer's sending direction. A TLS 1.2 `close_notify` closes both directions of the `SSLEngine`, so it
 ends the drain and accepted plaintext that has not been sent is discarded.
 
+Unless a clean shutdown is already waiting for the peer's reply, cancelling the transport output
+ends the session, because nothing more can be sent to the peer. That
+teardown is not silent: once the session is established, the operator still delivers the inbound
+data it can decrypt in the round that observes the cancellation, and then reports
+`SessionTruncated` before the plaintext output completes if the peer's `close_notify` has not been
+processed. This also applies with no buffered inbound records and during renegotiation of an
+established session: truncation means that a clean peer shutdown was not observed, rather than proof
+that application bytes were lost. Inbound records still buffered beyond that round are discarded.
+The operator does not wait for more ciphertext to complete a partial TLS record: cancellation can
+discard fragments still in flight, and the truncation signal makes that abandonment observable.
+An already pending plaintext element is delivered before the signal. A plaintext output that is
+already closed, or a session cancelled before its initial handshake finished, completes without that
+signal. The transport-output cancellation itself does not call `SSLEngine.closeInbound()` or
+invalidate the established `SSLSession`. If transport input has already completed and is depleted,
+the cancellation path skips the final inbound round and reports truncation directly.
+
+If the operator has already sent its own `close_notify` and is waiting for the peer's reply,
+transport-output cancellation preserves that existing wait while the plaintext output is open.
+If the local `close_notify` is still buffered inside the operator, cancellation discards it and
+follows the truncation behavior above instead of waiting for the peer's reply.
+Cancelling the plaintext output as well ends the wait. An operator still flushing outbound data
+cannot continue sending after transport-output cancellation.
+
+The truncation signal obeys plaintext backpressure. The plaintext input is cancelled immediately
+when the transport output is cancelled, even if the final inbound round is waiting for plaintext
+demand. The transport input is cancelled after that round; the operator then stays alive until
+the signal is consumed or the plaintext output is cancelled.
+Applications that stop reading must cancel that output as well; applications requiring a time bound
+can apply `completionTimeout` to the plaintext stream. Consumers using the TLS `BidiFlow` directly
+must handle `SessionTruncated` explicitly rather than using a partial `map` that only matches
+`SessionBytes`. The Scala and Java `Tcp.bindWithTls` and `Tcp.outgoingConnectionWithTls` APIs expose
+`ByteString` streams and filter out this signal.
+
 ## Streaming File IO
 
 Pekko Streams provide simple Sources and Sinks that can work with @apidoc[util.ByteString] instances to perform IO operations

diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala
@@ -37,7 +37,7 @@ import pekko.actor.Cancellable
 import pekko.event.LoggingAdapter
 import pekko.stream._
 import pekko.stream.TLSProtocol._
-import pekko.stream.impl.io.TlsGraphStage
+import pekko.stream.impl.io.{ TlsGraphStage, TlsModule }
 import pekko.stream.scaladsl._
 import pekko.stream.scaladsl.GraphDSL.Implicits._
 import pekko.stream.testkit.{ StreamSpec, TestPublisher, TestSubscriber }
@@ -310,16 +310,20 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
       negotiatedPacketSize: Int,
       negotiatedApplicationSize: Int,
       sessionProtocol: String = null,
-      wrapHandshakeOnNeedWrap: Boolean = false) extends SSLEngine {
+      wrapHandshakeOnNeedWrap: Boolean = false,
+      closeOnInitialWrap: Boolean = false) extends SSLEngine {
     private var handshakeStatus: SSLEngineResult.HandshakeStatus = NEED_UNWRAP
     @volatile var unwrapCount = 0
     private var inboundDone = false
-    private var outboundDone = false
+    @volatile private var outboundDone = false
     private var handshakePending = true
 
     @volatile var applicationWraps = 0
     @volatile var applicationBytes = 0
     @volatile var handshakesStarted = 0
+    @volatile var closeInboundCalls = 0
+    @volatile var unwrapFailure: SSLException = _
+    @volatile var failedUnwraps = 0
     @volatile var overflowWraps = 0
     @volatile var largestOutputCapacity = 0
     @volatile var inputArray: Array[Byte] = _
@@ -344,7 +348,7 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
     override def beginHandshake(): Unit = {
       handshakesStarted += 1
       handshakePending = true
-      handshakeStatus = NEED_UNWRAP
+      handshakeStatus = if (closeOnInitialWrap) NEED_WRAP else NEED_UNWRAP
     }
 
     override def getHandshakeStatus: SSLEngineResult.HandshakeStatus = handshakeStatus
@@ -360,6 +364,13 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
         overflowWraps += 1
         return new SSLEngineResult(BUFFER_OVERFLOW, handshakeStatus, 0, 0)
       }
+      if (closeOnInitialWrap && handshakePending) {
+        handshakePending = false
+        outboundDone = true
+        handshakeStatus = NOT_HANDSHAKING
+        dst.put(CloseCiphertext.head)
+        return new SSLEngineResult(CLOSED, NOT_HANDSHAKING, 0, 1)
+      }
       if (wrapHandshakeOnNeedWrap && handshakeStatus == NEED_WRAP && !(wrapOnCloseInbound && inboundDone)) {
         // Like the JDK, a wrap while NEED_WRAP produces only the handshake message and consumes no plaintext.
         // SunJSSE reports FINISHED for a TLS 1.3 KeyUpdate response, then NOT_HANDSHAKING from the engine.
@@ -399,6 +410,10 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
         dsts: Array[ByteBuffer],
         offset: Int,
         length: Int): SSLEngineResult = {
+      if (unwrapFailure ne null) {
+        failedUnwraps += 1
+        throw unwrapFailure
+      }
       if (src.hasArray) inputArray = src.array()
       if (!handshakePending && negotiatedPacketSize > 0) {
         reportedPacketSize = negotiatedPacketSize
@@ -437,6 +452,7 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
     }
 
     override def closeInbound(): Unit = {
+      closeInboundCalls += 1
       inboundDone = true
       handshakeStatus = if (wrapOnCloseInbound) NEED_WRAP else NOT_HANDSHAKING
     }
@@ -487,13 +503,24 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
       negotiatedPacketSize: Int = 0,
       negotiatedApplicationSize: Int = 0,
       sessionProtocol: String = null,
-      wrapHandshakeOnNeedWrap: Boolean = false): DeferredCloseHarness = {
+      wrapHandshakeOnNeedWrap: Boolean = false,
+      useLegacyActor: Boolean = false,
+      closeOnInitialWrap: Boolean = false): DeferredCloseHarness = {
     val engine =
       new ScriptedSSLEngine(TlsSpec.initSslContext("TLSv1.2").createSSLEngine(), wrapOnCloseInbound,
         requestHandshakeAfterApplication, requestWrapOnSubsequentUnwrap,
         requiredOutputSpace, negotiatedPacketSize, negotiatedApplicationSize, sessionProtocol,
-        wrapHandshakeOnNeedWrap)
+        wrapHandshakeOnNeedWrap, closeOnInitialWrap)
     val sessionVerified = Promise[Unit]()
+    val verifySession = (_: SSLSession) => {
+      sessionVerified.trySuccess(())
+      Success(())
+    }
+    val tlsFlow =
+      if (useLegacyActor)
+        BidiFlow.fromGraph(TlsModule(Attributes.none, () => engine, verifySession, closing))
+      else
+        BidiFlow.fromGraph(new TlsGraphStage(() => engine, verifySession, closing))
     val probes =
       RunnableGraph
         .fromGraph(
@@ -503,15 +530,7 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co
             TestSink[ByteString](),
             TestSink[SslTlsInbound]())(Tuple4.apply) { implicit b => (plainIn, cipherIn, cipherOut, plainOut) =>
             val tls = b.add(
-              BidiFlow
-                .fromGraph(new TlsGraphStage(
-                  () => engine,
-                  _ => {
-                    sessionVerified.trySuccess(())
-                    Success(())
-                  },
-                  closing))
-                .withAttributes(Attributes.inputBuffer(1, 1).and(extraAttributes)))
+              tlsFlow.withAttributes(Attributes.inputBuffer(1, 1).and(extraAttributes)))
             plainIn                                          ~> tls.in1
             tls.out1                                         ~> cipherOut
             cipherIn                                         ~> tls.in2
@@ -693,6 +712,125 @@ class TlsGraphStageDeferredCloseSpec extends TlsGraphStageBufferTestBase(TlsGrap
       harness.engine.applicationWraps shouldBe 0
     }
 
+    "complete without truncation when the transport output is cancelled before the initial handshake" in {
+      val harness = newHarness()
+
+      // No port has received an element or demand that would otherwise start the pump.
+      harness.cipherOut.cancel()
+
+      harness.plainIn.expectCancellation()
+      harness.cipherIn.expectCancellation()
+      harness.plainOut.expectComplete()
+      harness.engine.unwrapCount shouldBe 0
+      harness.engine.closeInboundCalls shouldBe 0
+    }
+
+    "report transport output cancellation without closing the established engine inbound" in {
+      val harness = newHarness()
+      finishInitialHandshake(harness)
+
+      harness.cipherOut.cancel()
+
+      harness.plainIn.expectCancellation()
+      harness.cipherIn.expectCancellation()
+      harness.plainOut.expectNext(SessionTruncated)
+      harness.plainOut.expectComplete()
+      harness.engine.closeInboundCalls shouldBe 0
+    }
+
+    "complete when transport output cancels before a deferred initial close is pushed" in {
+      assertAllStagesStopped {
+        val harness = newHarness(closeOnInitialWrap = true)
+        harness.cipherOut.request(1)
+        awaitAssert(harness.engine.isOutboundDone shouldBe true)
+        // The first output is held by InitialOutputFlushTimer until the explicit scheduler advances.
+        harness.cipherOut.expectNoMessage(100.millis)
+        harness.cipherOut.cancel()
+
+        harness.cipherIn.expectCancellation()
+        harness.plainIn.expectCancellation()
+        harness.plainOut.expectComplete()
+        harness.engine.unwrapCount shouldBe 0
+        harness.engine.closeInboundCalls shouldBe 0
+        harness.sessionVerified.future.isCompleted shouldBe false
+      }
+    }
+
+    "report transport output cancellation after the engine outbound has closed" in {
+      val harness = newHarness(closing = EagerClose)
+      finishInitialHandshake(harness)
+
+      harness.plainIn.sendComplete()
+      harness.cipherOut.request(1)
+      awaitAssert(harness.engine.isOutboundDone shouldBe true)
+      harness.cipherOut.cancel()
+
+      harness.cipherIn.expectCancellation()
+      harness.plainOut.expectNext(SessionTruncated)
+      harness.plainOut.expectComplete()
+      harness.engine.closeInboundCalls shouldBe 0
+    }
+
+    for {
+      useLegacyActor <- List(false, true)
+      bufferedInput <- List(false, true)
+    } {
+      val engine = if (useLegacyActor) "legacy actor" else "graph stage"
+      s"report cancellation after transport input completes without closing inbound ($engine, buffered=$bufferedInput)" in {
+        assertAllStagesStopped {
+          val harness = newHarness(
+            negotiatedPacketSize = TlsRecord.size,
+            negotiatedApplicationSize = 1,
+            useLegacyActor = useLegacyActor)
+          finishInitialHandshake(harness)
+
+          // Consume the handshake's remaining plaintext demand before completing transport input.
+          harness.cipherIn.sendNext(TlsRecord)
+          harness.plainOut.expectNext().asInstanceOf[SessionBytes].bytes shouldBe ByteString("A")
+          if (bufferedInput) harness.cipherIn.sendNext(TlsRecord)
+          // These probe calls enqueue completion, cancellation, then demand in that order.
+          harness.cipherIn.sendComplete()
+          harness.cipherOut.cancel()
+          // With nothing left to decrypt, teardown releases the plaintext input without waiting for demand.
+          if (!bufferedInput) harness.plainIn.expectCancellation()
+          harness.plainOut.request(2)
+
+          if (bufferedInput)
+            harness.plainOut.expectNext().asInstanceOf[SessionBytes].bytes shouldBe ByteString("A")
+          harness.plainOut.expectNext(SessionTruncated)
+          harness.plainOut.expectComplete()
+          if (bufferedInput) harness.plainIn.expectCancellation()
+          harness.engine.closeInboundCalls shouldBe 0
+        }
+      }
+    }
+
+    for (useLegacyActor <- List(false, true)) {
+      val engine = if (useLegacyActor) "legacy actor" else "graph stage"
+      s"preserve an unwrap failure during transport output cancellation ($engine)" in {
+        assertAllStagesStopped {
+          val harness = newHarness(
+            negotiatedPacketSize = TlsRecord.size,
+            negotiatedApplicationSize = 1,
+            useLegacyActor = useLegacyActor)
+          finishInitialHandshake(harness)
+          harness.cipherIn.sendNext(TlsRecord)
+          harness.plainOut.expectNext().asInstanceOf[SessionBytes].bytes shouldBe ByteString("A")
+
+          val failure = new SSLException("failed final inbound round")
+          harness.engine.unwrapFailure = failure
+          harness.cipherIn.sendNext(TlsRecord)
+          harness.cipherOut.cancel()
+          harness.plainOut.request(1)
+
+          harness.plainOut.expectError() shouldBe failure
+          harness.plainIn.expectCancellation()
+          harness.cipherIn.expectCancellation()
+          harness.engine.failedUnwraps shouldBe 1
+        }
+      }
+    }
+
     for (closing <- Seq(IgnoreComplete, EagerClose)) {
       s"drain accepted plaintext when transport input completes during cancellation with $closing" in {
         val harness = newHarness(closing = closing)