From 5a887c7a748c4af924dea61a548a56ae7319e29d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Thu, 29 Oct 2026 20:20:37 +0800 Subject: [PATCH] Preserve TLS plaintext ordering and drain on cancellation #3729 Motivation: Sender-side boundary batching could deliver plaintext-output cancellation before plaintext already delivered by the same handler. TLS could also close outbound before encrypting plaintext accepted into its deferred batch. Modification: - Opt TLS input boundaries out of sender-side batching while retaining asynchronous isolation and the existing input-buffer sizes. - Stop accepting input and drain the established session's accepted data with backpressure. Preserve transport-close ordering, failures during delayed cancellation, and prompt termination when a new handshake would be required. - Add deterministic ordering, deferred-close, and partially sent large-message regressions. Document drain limits in both DSLs and the guide, including the correct IgnoreCancel API reference. Result: Boundary messages sent by a handler precede its subsequent cancellation. Accepted plaintext drains when cancellation is observed before peer transport completion, subject to the documented limits. Preserve the peer-completes-first discard contract, original TCP hello assertion, and default closing policy. Tests: - Large-message cancellation regression for IgnoreComplete and EagerClose: 0/2 passed with the original main TlsGraphStage implementation, 2/2 with the current implementation. Deferred-close and cancellation suites: 19/19. - Previously recorded ordering regression on main: 0/2 before, 2/2 after; related TLS suites 252/252, fusing/interpreter suites 37/37, TCP TLS 20/20 per engine. The combined PR's original GraphStage TCP TLS case passed 1/1. - Temporary #3727/#3728 integration: all 15 original deferred-close cases and related TLS suites passed; another 20/20 TCP TLS runs per engine passed. - Native scalafmt and git diff --check passed for the new tests. Production code is unchanged by this test-only follow-up. - Existing docs/paradox, headerCreateAll and +headerCheckAll checks passed; documentation, imports, source-file headers, and public API are unchanged. - Auditable Qoder stdout review and independent review: no remaining must-fix findings after checking the test against the implementation and API definitions. - Full validatePullRequest and cross-project MiMa left to remote validation per user instruction; CI results were not followed or claimed to pass. References: Fixes #3729. Refs #2860. --- docs/src/main/paradox/stream/stream-io.md | 24 ++ .../io/TlsGraphStageCancellationSpec.scala | 110 ++++++++ .../stream/io/TlsGraphStageIsolatedSpec.scala | 237 +++++++++++++++++- .../impl/fusing/ActorGraphInterpreter.scala | 8 +- .../pekko/stream/impl/io/TlsGraphStage.scala | 71 +++++- .../org/apache/pekko/stream/javadsl/TLS.scala | 14 +- .../apache/pekko/stream/scaladsl/TLS.scala | 14 +- 7 files changed, 451 insertions(+), 27 deletions(-) create mode 100644 stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageCancellationSpec.scala diff --git a/docs/src/main/paradox/stream/stream-io.md b/docs/src/main/paradox/stream/stream-io.md index d5ed9cf4c7a..80537478401 100644 --- a/docs/src/main/paradox/stream/stream-io.md +++ b/docs/src/main/paradox/stream/stream-io.md @@ -161,6 +161,30 @@ Java The `SSLEngine` instance can then be used with the binding or outgoing connection factory methods. +The GraphStage TLS engine keeps an asynchronous boundary around TLS processing. It sends element +messages across that boundary without sender-side batching, preserving their order relative to a +later plaintext-output cancellation from the same handler. This orders boundary messages; it does +not drain elements still buffered in other operators or change TLS closing policies. + +### TLS closing policies + +The `closing` parameter of @apidoc[TLS](stream.*.TLS$) controls how plaintext-input completion +and plaintext-output cancellation close the connection; see @apidoc[TLSClosing](stream.TLSClosing). +When cancellation closes the connection, bytes not yet accepted by the operator may be discarded. +With `pekko.stream.materializer.tls.engine = graph-stage`, the operator stops further plaintext input +and attempts to send data already accepted for the established session, subject to transport +availability. This includes up to 64 queued messages plus plaintext already buffered for wrapping. +Message sizes are unbounded, so termination can be delayed until that data is sent. Data queued after +a renegotiation request is discarded, and draining stops if a new handshake is required. Cancellation +during a handshake or a transport failure can still prevent delivery. The default `legacy-actor` +engine discards accepted plaintext on cancellation. Use the `IgnoreCancel` closing policy +(`TLSClosing.ignoreCancel()` in Java) if the handler needs to continue producing a response after it +stops reading. + +The GraphStage drain also applies when plaintext-output cancellation is observed before the peer's +transport input completes. If transport-input completion is observed first, the existing inbound-close +contract applies and accepted plaintext is discarded, matching the legacy actor engine. + ## 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/TlsGraphStageCancellationSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageCancellationSpec.scala new file mode 100644 index 00000000000..6ee8d641a2b --- /dev/null +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageCancellationSpec.scala @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.pekko.stream.io + +import javax.net.ssl.SSLContext + +import scala.concurrent.duration._ +import scala.util.Success + +import org.apache.pekko +import pekko.stream._ +import pekko.stream.TLSProtocol._ +import pekko.stream.impl.io.TlsGraphStage +import pekko.stream.scaladsl._ +import pekko.stream.stage._ +import pekko.stream.testkit.StreamSpec +import pekko.stream.testkit.scaladsl.{ TestSink, TestSource } +import pekko.testkit.WithLogCapturing +import pekko.util.ByteString + +class TlsGraphStageCancellationSpec extends StreamSpec(TlsSpec.configOverrides) with WithLogCapturing { + + private def tls(context: SSLContext, client: Boolean) = + BidiFlow + .fromGraph(new TlsGraphStage( + () => { + val engine = context.createSSLEngine() + engine.setUseClientMode(client) + engine.setEnabledProtocols(Array(context.getProtocol)) + engine + }, + _ => Success(()), + IgnoreComplete)) + .withAttributes(TlsGraphStage.StreamTlsAttributes) + + private class EchoThenCancel extends GraphStage[FlowShape[ByteString, SslTlsOutbound]] { + val in = Inlet[ByteString]("EchoThenCancel.in") + val out = Outlet[SslTlsOutbound]("EchoThenCancel.out") + override val shape = FlowShape(in, out) + + override def createLogic(attributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) { + private var first = true + private var cancelOnPull = false + setHandler(in, + new InHandler { + override def onPush(): Unit = { + push(out, SendBytes(grab(in))) + if (first) first = false + else cancelOnPull = true + // Keep out open so its completion cannot flush the boundary batch before cancellation. + } + }) + setHandler(out, + new OutHandler { + override def onPull(): Unit = + if (cancelOnPull) { + // The boundary has consumed the preceding push and is asking for another element. + cancel(in) + } else pull(in) + }) + } + } + + "TlsGraphStage cancellation ordering" must { + List("TLSv1.2", "TLSv1.3").foreach { protocol => + s"encrypt plaintext pushed before cancelling its output with $protocol" in { + val context = TlsSpec.initSslContext(protocol) + // Bypass the separate small-write timer regression covered by the deferred-close tests. + val payload = ByteString("o" * (TlsGraphStage.MaxBatchedApplicationBytes + 1)) + val echo = Flow[SslTlsInbound] + .collect { case SessionBytes(_, bytes) => bytes } + .via(new EchoThenCancel) + + val (plainIn, plainOut) = TestSource[SslTlsOutbound]() + .via(tls(context, client = true).atop(tls(context, client = false).reversed).join(echo)) + .collect { case SessionBytes(_, bytes) => bytes } + // Exercise the normal interpreter ordering: push, next pull, cancellation, batch flush. + .withAttributes(ActorAttributes.fuzzingMode(false)) + .toMat(TestSink[ByteString]())(Keep.both) + .run() + + within(3.seconds) { + // Complete a round trip before testing cancellation, including TLS 1.3 handshake output. + plainOut.request(2) + plainIn.sendNext(SendBytes(ByteString("ready"))) + plainOut.expectNext(ByteString("ready")) + plainIn.sendNext(SendBytes(payload)) + plainOut.expectNext(payload) + plainOut.expectComplete() + plainIn.expectCancellation() + } + } + } + } +} 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 index 1ddc3d78b4f..544ff63fbe3 100644 --- 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 @@ -253,15 +253,24 @@ class TlsGraphStageDeferredCloseSpec extends StreamSpec(TlsGraphStageDeferredClo private val ApplicationCiphertext = ByteString(0x5A.toByte) private val CloseCiphertext = ByteString(0x5B.toByte) - private final class ScriptedSSLEngine(delegate: SSLEngine, wrapOnCloseInbound: Boolean) extends SSLEngine { + private final class ScriptedSSLEngine( + delegate: SSLEngine, + wrapOnCloseInbound: Boolean, + requestHandshakeAfterApplication: Boolean, + requestWrapOnSubsequentUnwrap: Boolean) extends SSLEngine { private var handshakeStatus: SSLEngineResult.HandshakeStatus = NEED_UNWRAP @volatile var unwrapCount = 0 private var inboundDone = false private var outboundDone = false @volatile var applicationWraps = 0 + @volatile var applicationBytes = 0 + @volatile var handshakesStarted = 0 - override def beginHandshake(): Unit = handshakeStatus = NEED_UNWRAP + override def beginHandshake(): Unit = { + handshakesStarted += 1 + handshakeStatus = NEED_UNWRAP + } override def getHandshakeStatus: SSLEngineResult.HandshakeStatus = handshakeStatus @@ -287,12 +296,14 @@ class TlsGraphStageDeferredCloseSpec extends StreamSpec(TlsGraphStageDeferredClo 1 } else if (consumed > 0) { applicationWraps += 1 + applicationBytes += consumed dst.put(ApplicationCiphertext.head) 1 } else 0 - handshakeStatus = NOT_HANDSHAKING - new SSLEngineResult(if (closeWrap) CLOSED else OK, NOT_HANDSHAKING, consumed, produced) + handshakeStatus = + if (consumed > 0 && requestHandshakeAfterApplication) NEED_UNWRAP else NOT_HANDSHAKING + new SSLEngineResult(if (closeWrap) CLOSED else OK, handshakeStatus, consumed, produced) } override def unwrap( @@ -307,9 +318,12 @@ class TlsGraphStageDeferredCloseSpec extends StreamSpec(TlsGraphStageDeferredClo if (unwrapCount == 1) { handshakeStatus = NOT_HANDSHAKING new SSLEngineResult(OK, FINISHED, consumed, 0) - } else { + } else if (requestWrapOnSubsequentUnwrap) { handshakeStatus = NEED_WRAP new SSLEngineResult(OK, NEED_WRAP, consumed, 0) + } else { + handshakeStatus = NOT_HANDSHAKING + new SSLEngineResult(OK, NOT_HANDSHAKING, consumed, 0) } } @@ -350,9 +364,15 @@ class TlsGraphStageDeferredCloseSpec extends StreamSpec(TlsGraphStageDeferredClo val cipherOut: TestSubscriber.Probe[ByteString], val plainOut: TestSubscriber.Probe[SslTlsInbound]) - private def newHarness(wrapOnCloseInbound: Boolean = false): DeferredCloseHarness = { + private def newHarness( + wrapOnCloseInbound: Boolean = false, + closing: TLSClosing = IgnoreComplete, + requestHandshakeAfterApplication: Boolean = false, + requestWrapOnSubsequentUnwrap: Boolean = true, + extraAttributes: Attributes = Attributes.none): DeferredCloseHarness = { val engine = - new ScriptedSSLEngine(TlsSpec.initSslContext("TLSv1.2").createSSLEngine(), wrapOnCloseInbound) + new ScriptedSSLEngine(TlsSpec.initSslContext("TLSv1.2").createSSLEngine(), wrapOnCloseInbound, + requestHandshakeAfterApplication, requestWrapOnSubsequentUnwrap) val probes = RunnableGraph .fromGraph( @@ -363,12 +383,16 @@ class TlsGraphStageDeferredCloseSpec extends StreamSpec(TlsGraphStageDeferredClo TestSink[SslTlsInbound]())(Tuple4.apply) { implicit b => (plainIn, cipherIn, cipherOut, plainOut) => val tls = b.add( BidiFlow - .fromGraph(new TlsGraphStage(() => engine, _ => Success(()), IgnoreComplete)) - .withAttributes(Attributes.inputBuffer(1, 1))) - plainIn ~> tls.in1 - tls.out1 ~> cipherOut - cipherIn ~> tls.in2 - tls.out2 ~> plainOut + .fromGraph(new TlsGraphStage(() => engine, _ => Success(()), closing)) + .withAttributes(Attributes.inputBuffer(1, 1).and(extraAttributes))) + plainIn ~> tls.in1 + tls.out1 ~> cipherOut + cipherIn ~> tls.in2 + if (extraAttributes == Attributes.none) tls.out2 ~> plainOut + else { + // Keep the subscriber boundary's cancellation independent of the TLS stage's attributes. + tls.out2 ~> Flow[SslTlsInbound].map(identity) ~> plainOut + } ClosedShape }) .run() @@ -389,6 +413,193 @@ class TlsGraphStageDeferredCloseSpec extends StreamSpec(TlsGraphStageDeferredClo } "TlsGraphStage deferred close handling" must { + + for (closing <- Seq(IgnoreComplete, EagerClose)) { + s"flush accepted plaintext on output cancellation with $closing before the batch timer fires" in { + val harness = newHarness(closing = closing) + finishInitialHandshake(harness) + + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() // the stage owns the bytes, but the batch timer has not run + harness.plainOut.cancel() + harness.plainIn.expectCancellation() + + // Cancellation must stop new input without discarding the accepted batch or bypassing backpressure. + harness.engine.applicationWraps shouldBe 0 + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + harness.cipherOut.expectComplete() + harness.cipherIn.expectCancellation() + harness.engine.applicationBytes shouldBe 1 + } + + s"drain a partially sent large message on output cancellation with $closing" in { + val harness = newHarness(closing = closing) + finishInitialHandshake(harness) + // The scripted engine consumes one input buffer per wrap. Keep a tail in the accepted ByteString + // so draining must refill that buffer on later transport requests; TLS record boundaries are not modeled. + val applicationBuffers = TlsGraphStage.MaxApplicationRecordsPerEngineCall * 2 + 1 + val payload = ByteString("a" * (harness.engine.getSession.getApplicationBufferSize * applicationBuffers)) + harness.plainIn.sendNext(SendBytes(payload)) + harness.plainIn.expectRequest() + + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + val initiallySent = harness.engine.applicationBytes + initiallySent should be > 0 + initiallySent should be < payload.size + + harness.plainOut.cancel() + harness.plainIn.expectCancellation() + // Processing cancellation must not wrap more plaintext without transport demand. + harness.engine.applicationBytes shouldBe initiallySent + + var sent = initiallySent + while (sent < payload.size) { + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + val nextSent = harness.engine.applicationBytes + nextSent should be > sent + nextSent should be <= payload.size + sent = nextSent + } + // The last data element consumed the demand; allow the outbound close to run as well. + harness.cipherOut.request(1) + harness.cipherOut.expectComplete() + harness.cipherIn.expectCancellation() + harness.engine.applicationBytes shouldBe payload.size + } + } + + "close without waiting for peer handshake input while draining cancelled plaintext" in { + val harness = newHarness(requestHandshakeAfterApplication = true) + finishInitialHandshake(harness) + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + harness.plainOut.cancel() + harness.plainIn.expectCancellation() + + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + harness.cipherOut.expectComplete() + harness.cipherIn.expectCancellation() + harness.engine.applicationBytes shouldBe 1 + } + + "stop draining when the transport output is cancelled" in { + val harness = newHarness() + finishInitialHandshake(harness) + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + harness.plainOut.cancel() + harness.plainIn.expectCancellation() + + harness.cipherOut.cancel() + harness.cipherIn.expectCancellation() + harness.engine.applicationWraps shouldBe 0 + } + + for (closing <- Seq(IgnoreComplete, EagerClose)) { + s"drain accepted plaintext when transport input completes during cancellation with $closing" in { + val harness = newHarness(closing = closing) + finishInitialHandshake(harness) + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + harness.plainOut.cancel() + harness.plainIn.expectCancellation() + harness.cipherIn.sendComplete() + + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + harness.cipherOut.expectComplete() + harness.engine.applicationBytes shouldBe 1 + } + } + + for (closing <- Seq(IgnoreCancel, IgnoreBoth)) { + s"terminate when both outputs are cancelled with $closing and silent transport input" in { + val harness = newHarness(closing = closing) + finishInitialHandshake(harness) + harness.plainOut.cancel() + harness.cipherOut.cancel() + + harness.cipherIn.expectCancellation() + harness.plainIn.expectCancellation() + } + } + + "fail transport when upstream fails during delayed cancellation" in { + val harness = newHarness(requestWrapOnSubsequentUnwrap = false, + extraAttributes = Attributes( + Attributes.CancellationStrategy( + Attributes.CancellationStrategy.AfterDelay(1.second, Attributes.CancellationStrategy.PropagateFailure)))) + finishInitialHandshake(harness) + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + harness.plainOut.cancel() + // A second signal from the cancelled output proves that its cancellation was processed. + harness.cipherIn.sendNext(TlsRecord) + harness.cipherIn.expectRequest() + awaitAssert(harness.engine.unwrapCount shouldBe 2) + val failure = new RuntimeException("plaintext input failed during cancellation") + harness.plainIn.sendError(failure) + + harness.cipherOut.expectError() shouldBe failure + system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler].timePasses(1.second) + harness.cipherIn.expectCancellation() + } + + "preserve the accepted batch when upstream completes during delayed cancellation" in { + val harness = newHarness(extraAttributes = Attributes( + Attributes.CancellationStrategy( + Attributes.CancellationStrategy.AfterDelay(1.second, Attributes.CancellationStrategy.PropagateFailure)))) + finishInitialHandshake(harness) + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + + harness.plainOut.cancel() + harness.plainIn.sendComplete() + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + harness.cipherOut.expectComplete() + harness.engine.applicationBytes shouldBe 1 + system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler].timePasses(1.second) + harness.cipherIn.expectCancellation() + } + + "close promptly on output cancellation during the initial handshake" in { + val harness = newHarness() + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + harness.plainOut.cancel() + + harness.plainIn.expectCancellation() + harness.cipherIn.expectCancellation() + harness.cipherOut.expectComplete() + harness.engine.applicationWraps shouldBe 0 + } + + "drain only the current session before a queued renegotiation on output cancellation" in { + val harness = newHarness() + finishInitialHandshake(harness) + + harness.plainIn.sendNext(SendBytes(ByteString("a"))) + harness.plainIn.expectRequest() + harness.plainIn.sendNext(NegotiateNewSession) + harness.plainIn.expectRequest() + harness.plainIn.sendNext(SendBytes(ByteString("b"))) + harness.plainIn.expectRequest() + harness.plainOut.cancel() + harness.plainIn.expectCancellation() + + harness.cipherOut.request(1) + harness.cipherOut.expectNext(ApplicationCiphertext) + harness.cipherOut.expectComplete() + harness.cipherIn.expectCancellation() + harness.engine.applicationBytes shouldBe 1 + harness.engine.handshakesStarted shouldBe 1 + } + "complete when inbound closes before a deferred user batch runs" in { val harness = newHarness() finishInitialHandshake(harness) diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala index b9cea1a38ac..b91d73e7332 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala @@ -57,6 +57,9 @@ import org.reactivestreams.Subscription def batch(elements: Array[AnyRef]): Unit } + // Opt out of sender-side batching when ordering with signals on another port matters (TLS). + private[stream] case object UnbatchedInput extends Attributes.Attribute + sealed abstract class BoundaryEvent extends DeadLetterSuppression with NoSerializationVerificationNeeded { def shell: GraphInterpreterShell @@ -560,8 +563,9 @@ import org.reactivestreams.Subscription if (subscriber eq null) { subscriber = sub boundarySubscriber = sub match { - case boundary: BoundarySubscriber => boundary - case _ => null + case boundary: BatchingActorInputBoundary if boundary.attributes.contains(UnbatchedInput) => null + case boundary: BoundarySubscriber => boundary + case _ => null } val subscription = new Subscription with SubscriptionWithCancelException { override def request(elements: Long): Unit = actor ! new RequestMore(ActorOutputBoundary.this, elements) diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala index 34331c7410a..79a5dd5ea91 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala @@ -34,6 +34,7 @@ import pekko.io.{ BufferPool, Tcp } import pekko.stream._ import pekko.stream.TLSProtocol._ import pekko.stream.impl.Stages.DefaultAttributes +import pekko.stream.impl.fusing.ActorGraphInterpreter import pekko.stream.stage.{ GraphStage, InHandler, OutHandler, TimerGraphStageLogic } import pekko.util.ByteString @@ -189,8 +190,9 @@ import pekko.util.ByteString case FlushingOutboundPhase => outboundHalfClosedReady && !outboundHalfClosedCompleted case AwaitingClosePhase => awaitingCloseReady && !awaitingCloseCompleted case OutboundClosedPhase => (outboundHalfClosedReady || inboundReady) && !outboundClosedCompleted - case InboundClosedPhase => (outboundReady || inboundHalfClosedReady) && !inboundClosedCompleted - case _ => false + case InboundClosedPhase => + (outboundReady || inboundHalfClosedReady || mustCloseOutboundOnCancel) && !inboundClosedCompleted + case _ => false } private def phaseCompleted: Boolean = @@ -278,7 +280,7 @@ import pekko.util.ByteString engineNeedsWrapCompleted || transportOutput.isClosed private def inboundHalfClosedReady: Boolean = - transportInput.isReady && !engine.isInboundDone + transportInput.isReady && !engine.isInboundDone && !drainBeforeInboundClose private def inboundHalfClosedCompleted: Boolean = transportInput.isCompleted || engine.isInboundDone @@ -286,8 +288,8 @@ import pekko.util.ByteString private def outboundClosedCompleted: Boolean = outboundHalfClosedCompleted && inboundCompleted - private def inboundClosedCompleted: Boolean = - outboundCompleted && inboundHalfClosedCompleted + private def inboundClosedCompleted: Boolean = transportOutput.isClosed || + (outboundCompleted && inboundHalfClosedCompleted) private def awaitingCloseReady: Boolean = transportInput.isPending && !engine.isInboundDone @@ -310,7 +312,7 @@ import pekko.util.ByteString } private def doInbound(isOutboundClosed: Boolean, inboundHalfClosedMode: Boolean): Boolean = - if (transportInput.isDepleted && transportInput.isEmpty) { + if (transportInput.isDepleted && transportInput.isEmpty && !drainBeforeInboundClose) { try engine.closeInbound() catch { case _: SSLException => userOutput.enqueue(SessionTruncated) @@ -321,12 +323,24 @@ import pekko.util.ByteString } else if (!inboundHalfClosedMode && userOutput.isCancelled) { if (!isOutboundClosed && closing.ignoreCancel) { nextPhase(InboundClosedPhase) + true + } else if (!isOutboundClosed && canDrainAcceptedUserInput && + (userInBufferHasRemaining || !userInput.isEmpty)) { + userInput.finishAcceptedInput() + nextPhase(InboundClosedPhase) + false } else { engine.closeOutbound() lastHandshakeStatus = engine.getHandshakeStatus nextPhase(FlushingOutboundPhase) + true } - true + } else if (inboundHalfClosedMode && mustCloseOutboundOnCancel) { + // Cancellation must not wait for a new handshake in order to drain application data. + engine.closeOutbound() + lastHandshakeStatus = engine.getHandshakeStatus + nextPhase(FlushingOutboundPhase) + false } else if (if (inboundHalfClosedMode) inboundHalfClosedReady else inboundReady) { transportInput.chopInto(transportInBuffer) if (TlsEngineHelpers.hasCompleteTlsRecord(transportInBuffer)) @@ -370,6 +384,15 @@ import pekko.util.ByteString } } + private def mustCloseOutboundOnCancel: Boolean = + !closing.ignoreCancel && !canDrainAcceptedUserInput + + private def drainBeforeInboundClose: Boolean = + transportInput.isDepleted && transportInput.isEmpty && userOutput.isCancelled && + !closing.ignoreCancel && canDrainAcceptedUserInput && (userInBufferHasRemaining || !userInput.isEmpty) + + private def canDrainAcceptedUserInput: Boolean = (stateBits & PlainDataAllowedFlag) != 0 && mayCloseOutbound + private def mayCloseOutbound: Boolean = lastHandshakeStatus match { case HandshakeStatus.NOT_HANDSHAKING | HandshakeStatus.FINISHED => true @@ -789,6 +812,36 @@ import pekko.util.ByteString !hasBeenPulled(inlet)) pull(inlet) + def finishAcceptedInput(): Unit = { + cancelPendingBatchTimer() + if (!logic.isClosed(inlet)) logic.cancel(inlet) + // AfterDelay cancellation leaves the inlet open temporarily; completion must not abort the drain. + setHandler(inlet, + new InHandler { + override def onPush(): Unit = () + override def onUpstreamFinish(): Unit = () + override def onUpstreamFailure(ex: Throwable): Unit = failTls(ex) + }) + stateBits |= UserInputFinishedFlag + + // Only drain data belonging to the current session. Do not start a queued renegotiation + // or send the data following it in the old session while closing. + var retained = 0 + while (retained < pendingCount && + pendingQueue((pendingHead + retained) & UserInputQueueMask).isInstanceOf[SendBytes]) + retained += 1 + var index = retained + while (index < pendingCount) { + pendingQueue((pendingHead + index) & UserInputQueueMask) = null + index += 1 + } + pendingCount = retained + if (pendingCount == 0) { + pendingHead = 0 + stateBits &= ~UserInputHasPendingFlag + } + } + def cancel(): Unit = if ((stateBits & UserInputCancelledFlag) == 0) { stateBits |= UserInputCancelledFlag @@ -1081,7 +1134,9 @@ import pekko.util.ByteString private[pekko] final val UserInputQueueSize = InputBufferMaxSize private[pekko] final val UserInputQueueMask = UserInputQueueSize - 1 private[pekko] val StreamTlsAttributes: Attributes = - Attributes.asyncBoundary and Attributes.inputBuffer(InputBufferInitialSize, InputBufferMaxSize) + Attributes.asyncBoundary and Attributes.inputBuffer(InputBufferInitialSize, InputBufferMaxSize) and + // Send plaintext messages before a subsequent plaintext-output cancellation from the same handler. + Attributes(ActorGraphInterpreter.UnbatchedInput) private[pekko] final val PlainDataAllowedFlag = 1 << 0 private[pekko] final val StageClosedFlag = 1 << 1 diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/TLS.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/TLS.scala index 67346ec1f65..259a42f531d 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/TLS.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/TLS.scala @@ -46,8 +46,18 @@ import pekko.util.ByteString * that it transports—to be precise a half-close will always promptly lead to a * full close. This means that canceling the plaintext output or completing the * plaintext input of the SslTls operator will lead to full termination of the - * secure connection without regard to whether bytes are remaining to be sent or - * received, respectively. Especially for a client the common idiom of attaching + * secure connection, subject to the closing policy described below. Bytes not yet + * accepted by the operator may be discarded. When cancellation closes the + * connection, the `legacy-actor` engine also discards accepted plaintext. The + * `graph-stage` engine stops accepting plaintext and attempts to send data already + * accepted for the established session, subject to transport availability. This + * drain includes up to 64 queued messages plus any plaintext already buffered for + * wrapping; message sizes are unbounded, so cancellation can delay termination + * until that data is sent. Data queued after a renegotiation request is discarded, + * and draining stops if a new handshake is required. The engine is selected with + * `pekko.stream.materializer.tls.engine`, which defaults to `legacy-actor`. + * + * Especially for a client the common idiom of attaching * a finite Source to the plaintext input and transforming the plaintext response * bytes coming out will not work out of the box due to early termination of the * connection. For this reason there is a parameter that determines whether the diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala index ce15dc94c08..c95d43c5489 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala @@ -48,8 +48,18 @@ import com.typesafe.config.{ ConfigException, ConfigFactory } * that it transports—to be precise a half-close will always promptly lead to a * full close. This means that canceling the plaintext output or completing the * plaintext input of the SslTls operator will lead to full termination of the - * secure connection without regard to whether bytes are remaining to be sent or - * received, respectively. Especially for a client the common idiom of attaching + * secure connection, subject to the closing policy described below. Bytes not yet + * accepted by the operator may be discarded. When cancellation closes the + * connection, the `legacy-actor` engine also discards accepted plaintext. The + * `graph-stage` engine stops accepting plaintext and attempts to send data already + * accepted for the established session, subject to transport availability. This + * drain includes up to 64 queued messages plus any plaintext already buffered for + * wrapping; message sizes are unbounded, so cancellation can delay termination + * until that data is sent. Data queued after a renegotiation request is discarded, + * and draining stops if a new handshake is required. The engine is selected with + * `pekko.stream.materializer.tls.engine`, which defaults to `legacy-actor`. + * + * Especially for a client the common idiom of attaching * a finite Source to the plaintext input and transforming the plaintext response * bytes coming out will not work out of the box due to early termination of the * connection. For this reason there is a parameter that determines whether the