He-Pin commented on code in PR #3727:
URL: https://github.com/apache/pekko/pull/3727#discussion_r4408559696
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -321,12 +322,24 @@ import pekko.util.ByteString
} else if (!inboundHalfClosedMode && userOutput.isCancelled) {
if (!isOutboundClosed && closing.ignoreCancel) {
nextPhase(InboundClosedPhase)
+ true
+ } else if (!isOutboundClosed && canDrainAcceptedUserInput &&
Review Comment:
The drain can be preempted by the `transportInput.isDepleted &&
transportInput.isEmpty` branch at the top of `doInbound` (line 314), which is
evaluated *before* this branch and before the `inboundHalfClosedMode &&
cancelledHandshakeReady` branch below it.
Once we are in `InboundClosedPhase`, if the peer's transport input completes
while the accepted batch is still waiting for `cipherOut` demand, that first
branch fires: `engine.closeInbound()` -> `completeOrFlush()` ->
`FlushingOutboundPhase`/`CompletedPhase`, and the retained `SendBytes` are then
dropped by `pumpFinished()` -> `userInput.cancel()` -> `clearPending()`.
Reproduced against this PR's head with a probe added to
`TlsGraphStageDeferredCloseSpec`:
```scala
val harness = newHarness()
finishInitialHandshake(harness)
harness.plainIn.sendNext(SendBytes(ByteString("a")))
harness.plainIn.expectRequest()
harness.plainOut.cancel()
harness.plainIn.expectCancellation()
harness.cipherIn.sendComplete() // peer half-closes during the drain
harness.cipherOut.request(1)
harness.cipherOut.expectNext(ApplicationCiphertext)
```
```
[info] - must PROBE keep draining accepted plaintext when the transport
input completes *** FAILED *** (82 milliseconds)
[info] java.lang.AssertionError: assertion failed: expected
OnNext(ByteString(90)), found OnComplete
```
This is not a regression (`main` drops the data too, and so does
`TLSActor`), but it defeats the stated goal of the PR in a realistic ordering:
a client that sends a request and then half-closes its write side, or a server
whose `cipherIn` FIN is processed just before the handler cancels `plainOut`.
`TcpSpec`'s echo case only passes because `Source.maybe` keeps the client's
plaintext input open forever.
Suggested direction: while `canDrainAcceptedUserInput` and user data is
still queued, do the inbound close *after* the outbound drain - e.g. record
"transport depleted" and stay in `InboundClosedPhase` so `doOutbound` can wrap
the retained batch first, instead of jumping straight to
`FlushingOutboundPhase`/`CompletedPhase`. Please add a regression test for this
ordering either way.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -286,8 +287,8 @@ import pekko.util.ByteString
private def outboundClosedCompleted: Boolean =
outboundHalfClosedCompleted && inboundCompleted
- private def inboundClosedCompleted: Boolean =
- outboundCompleted && inboundHalfClosedCompleted
+ private def inboundClosedCompleted: Boolean = (!closing.ignoreCancel &&
transportOutput.isCancelled) ||
Review Comment:
This new escape hatch is guarded by `!closing.ignoreCancel`, so it only
fixes non-termination for `EagerClose`/`IgnoreComplete`. The identical hang
still exists for `IgnoreCancel` and `IgnoreBoth`.
With `closing = IgnoreCancel`, cancelling `plainOut` takes the first branch
of `doInbound` -> `InboundClosedPhase`. If `cipherOut` is then cancelled while
`cipherIn` stays open and silent: `outboundReady` is false
(`transportOutput.isDemandAvailable` is false once cancelled),
`inboundHalfClosedReady` is false (no transport data), and
`cancelledHandshakeReady` is false because `closing.ignoreCancel` is true - so
the phase is not executable. And `inboundClosedCompleted` stays false because
`inboundHalfClosedCompleted` needs `transportInput.isCompleted ||
engine.isInboundDone`. `pumpFinished()` is never reached and the stage lives
forever.
Reproduced on both `main` and this PR's head, so it is pre-existing rather
than introduced here:
```scala
val harness = newHarness(closing = IgnoreCancel)
finishInitialHandshake(harness)
harness.plainOut.cancel()
harness.cipherOut.cancel()
harness.cipherIn.expectCancellation() // never arrives
```
```
main : [info] Tests: succeeded 3, failed 1
pr3727 : [info] Tests: succeeded 10, failed 2
[info] - must PROBE terminate when the transport output is cancelled under
IgnoreCancel *** FAILED *** (3 seconds, 190 milliseconds)
[info] java.lang.AssertionError: assertion failed: timeout (3 seconds)
during fishForSpecificMessage, hint: Expecting cancellation
```
Since this line is already being touched, consider dropping the guard and
widening to `isClosed` (which also covers the errored/completed flags):
```scala
private def inboundClosedCompleted: Boolean = transportOutput.isClosed ||
(outboundCompleted && inboundHalfClosedCompleted)
```
If `cipherOut` is gone there is nowhere left to write, so terminating is
correct regardless of the closing policy. Happy for this to be a separate PR if
you would rather keep this one scoped.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -789,6 +807,31 @@ 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, ignoreTerminateInput)
Review Comment:
`ignoreTerminateInput` ignores `onPush`, `onUpstreamFinish` **and**
`onUpstreamFailure`.
Ignoring completion is the point of the change, but ignoring failure is a
behaviour change: today `UserInputSlot.onUpstreamFailure` routes to
`failTls(ex)`, which fails `cipherOut` and aborts the connection. With
`CancellationStrategy.AfterDelay` the inlet is genuinely still open here -
`GraphStageLogic.cancel` installs `EagerTerminateInput` precisely so that
termination signals are *not* ignored (see the `do not ignore termination
signals` comment at `GraphStage.scala:777-781`), and this line replaces it. The
net effect is that an upstream failure arriving during the drain is silently
converted into a clean `close_notify`.
Suggest a small private handler that ignores `onPush`/`onUpstreamFinish` but
still calls `failTls` on `onUpstreamFailure`.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -370,6 +383,11 @@ import pekko.util.ByteString
}
}
+ private def cancelledHandshakeReady: Boolean =
Review Comment:
Nit: `cancelledHandshakeReady` reads as "a cancelled handshake is ready",
but it actually means "cancellation must close outbound now because draining is
not possible". Something like `mustCloseOutboundOnCancel` would make both call
sites (`phaseExecutable` and `doInbound`) self-explanatory.
##########
docs/src/main/paradox/stream/stream-io.md:
##########
@@ -161,6 +161,14 @@ Java
The `SSLEngine` instance can then be used with the binding or outgoing
connection factory methods.
+When the closing policy makes plaintext-output cancellation close the
connection, the GraphStage TLS
Review Comment:
Two notes:
- This section is about constructing the `SSLEngine`; closing-policy
semantics would sit better next to the `TLSClosing` / `TLS()` documentation,
and the paragraph never mentions `pekko.stream.materializer.tls.engine`, so a
reader cannot tell whether it applies to them.
- #3728 inserts a different paragraph at this exact position, so whichever
PR merges second will conflict here and both paragraphs need to be kept.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala:
##########
@@ -363,12 +378,13 @@ 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)))
+ .fromGraph(new TlsGraphStage(() => engine, _ => Success(()),
closing))
+ .withAttributes(Attributes.inputBuffer(1,
1).and(extraAttributes)))
plainIn ~> tls.in1
tls.out1 ~> cipherOut
cipherIn ~> tls.in2
- tls.out2 ~> plainOut
+ // Keep the subscriber boundary's cancellation independent of the
TLS stage's attributes.
+ tls.out2 ~> Flow[SslTlsInbound].map(identity) ~> plainOut
Review Comment:
This `map(identity)` is only needed for the `AfterDelay` case, so that the
TestSink's own cancellation is not subject to the TLS stage's attributes - but
it is now inserted into every harness, which changes the topology of the three
pre-existing tests as well. Consider making it conditional on
`extraAttributes`, or extending the comment to say that it is deliberate for
all cases.
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala:
##########
@@ -48,8 +48,15 @@ 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. Pending
bytes
+ * may be discarded. When cancellation closes the connection, the
`legacy-actor`
Review Comment:
Two problems with this wording (the same text is added to
`javadsl/TLS.scala`):
1. `Pending bytes may be discarded.` is immediately contradicted by the
following sentences, which say the `graph-stage` engine *does* send
already-accepted data. As written, a reader cannot tell what actually happens.
2. The drain bound is not stated. `finishAcceptedInput()` stops new input
but drains everything already accepted, which is up to `UserInputQueueSize =
64` queued messages (`TlsGraphStage.scala:1081`) plus whatever is in
`userInBuffer`, each of arbitrary size. The paragraph further down - "In order
to terminate the connection the client will then need to cancel the plaintext
output as soon as all expected bytes have been received" - now implies
termination can be delayed by that drain, which is worth saying explicitly.
Also, `graph-stage` is not the default (`reference.conf`:
`pekko.stream.materializer.tls.engine = "legacy-actor"`), and nothing here
tells the reader which engine they are on or where to look. Consider naming the
config key, the way the companion paragraph added in #3728 does.
Suggested shape: state the guarantee once ("bytes that have not been
accepted by the operator may be discarded"), then the engine-specific drain
behaviour and its bound, then point at `pekko.stream.materializer.tls.engine`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
