Websocket stream failure on receive/send timeout (network lost detection) #4289

gawronA · 2023-07-28T15:29:10Z

Hello all,

I would like to contribute to Akka project by providing a feature that allows to catch websocket disconnections due to network unavailability.
I have introduced two configuration parameters for websocket client:

akka.http.client.websocket.receive-idle-timeout
akka.http.client.websocket.send-idle-timeout

and server:

akka.http.server.websocket.receive-idle-timeout
akka.http.server.websocket.send-idle-timeout

that can be configured together with akka.http.{client, server}.periodic-keep-alive-max-idle to detect lost network connection.

The working principle is simple: if elements are not send or received (including ping/pong frames), the streams are failed with
akka.stream.StreamIdleTimeoutException

The default config values are infinite which means that this feature is disabled.

I have provided two new test modules: WebSocketServerReceiveIdleTimeoutSpec.scala and WebSocketServerSendIdleTimeoutSpec.scala because I couldn't find out how config overrides can be injected upon server creation (they are always taken from ActorSystem settings). If anyone have a better idea how this can be resolved, I'm open to proposals.

Thanks!

References #2419

* Add receiveIdleTimeout and sendIdleTimeout config parameters to websocket client and server * Add idleTimeout stack layer * Add integration tests

lightbend-cla-validator · 2023-07-28T15:29:17Z

Hi @gawronA,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Lightbend Contributors License Agreement:

https://www.lightbend.com/contribute/cla

gawronA · 2023-07-28T15:37:41Z

CLA signed.

johanandren

Seems like a nice/useful feature, impl looks pretty good, I left some suggestions.

akka-http-core/src/main/scala/akka/http/impl/engine/ws/WebSocket.scala

...ttp-core/src/test/scala/akka/http/impl/engine/ws/WebSocketServerReceiveIdleTimeoutSpec.scala

* Use BidiFlow.identity for default case instead of BidiFlow constructed from two identity Flows * Remove redundant maps * Add exception class matchers in tests

gawronA · 2023-08-15T21:24:01Z

@johanandren thank you for the review. I applied your suggestions.
Meanwhile I've found one issue with a case where akka.http.client.websocket.send-idle-timeout is set to non-infinite value and timeout is issued. Then the exception class that we get is akka.stream.scaladsl.TcpIdleTimeoutException instead of akka.stream.StreamIdleTimeoutException. This is caused by Akka stream tcp stage where any exceptions derived from java.util.concurrent.TimeoutException (including akka.stream.StreamIdleTimeoutException) are mapped to TcpIdleTimeoutException.
My proposals to resolve it:

Intentionally disable idle-timeout handling by tcp layer and prepare implementation on http layer like in http2 server
Leave it like that and properly document the case that when akka.http.client.websocket.send-idle-timeout is configured, the handler stream will be failed with akka.stream.scaladsl.TcpIdleTimeoutException
Remove the akka.http.client.websocket.send-idle-timeout config parameter and related functionality

Also feel free to directly push whatever headers are needed to WebSocketServerReceiveIdleTimeoutSpec.scala and WebScoketServerSendIdleTimeoutSpec.scala.

johanandren

Do a sbt "headerCreateAll; compile" to get the headers added and the code formatted and check in the result.

I think the TCP stage turning any stream idle timeout to TCP timeout is unfortunate but something that we could possibly fix in Akka core instead of doing anything about here.

johanandren · 2023-08-16T07:02:58Z

akka-http-core/src/test/scala/akka/http/impl/engine/ws/WebSocketIntegrationSpec.scala

+            Thread.sleep(200)
+          }, Duration.apply(3, TimeUnit.SECONDS))
+          x
+        })


Replace the in-stream-blocking map+sleep in the tests with .delay(200.millis)

gawronA · 2023-08-18T20:10:00Z

In the tests I want to simulate constant flow of messages in one direction and ensure that we still get timeout when messages are not sent in opposite direction.
I managed to replace the in-stream-blocking map with .delay(200.mililis) in send-idle-timeout test, but when I also changed it in receive-idle-timeout test it failed cause the stream wasn't time-outed. This is something I didn't expect to happen and investigated that this piece of code:

val future = Source(1 to 10).map(_ => TextMessage("dummy"))
  .map(x => {
    Await.result(Future {
      Thread.sleep(200)
    }, 3.seconds)
    x
  })
  .via(Http().webSocketClientFlow(
    WebSocketRequest("ws://127.0.01:" + myPort),
    settings = ClientConnectionSettings("akka.http.client.websocket.receive-idle-timeout = 1s"))
  ).toMat(Sink.foreach(println))(Keep.right).run()

produces no output - expected. However, this piece of code (map replaced with delay):

val future = Source(1 to 10).map(_ => TextMessage("dummy"))
  .delay(300.millis, DelayOverflowStrategy.backpressure).addAttributes(Attributes.inputBuffer(1, 1))
  .via(Http().webSocketClientFlow(
    WebSocketRequest("ws://127.0.01:" + myPort),
    settings = ClientConnectionSettings("akka.http.client.websocket.receive-idle-timeout = 1s"))
  ).toMat(Sink.foreach(println))(Keep.right).run()

produces:

TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)
TextMessage.Strict(dummy)

which is quite unexpected as the .delay operator somehow passes elements through the websocket flow (as if they were received from server) and therefore the timeout don't happen.

I may be wrong but in my opinion the .delay operator or inputBuffer attribute somehow interfere websocket flow.

johanandren · 2023-08-21T08:11:29Z

akka-http-core/src/test/scala/akka/http/impl/engine/ws/WebSocketIntegrationSpec.scala

+
+    "fail the materialized future with akka.stream.StreamIdleTimeoutException if elements are not received within receive-idle-timeout" in {
+      val bindingFuture = Http().newServerAt("localhost", 0).bindSync({
+        _.attribute(webSocketUpgrade).get.handleMessages(Flow.apply, None)


Flow.apply == Flow.identity so this is an echo-server, the test expects it to be a server that does not emit anything, something like Flow.fromSinkAndSourceCoupled(Sink.ignore, Source.maybe[Message]) could work in a test (in real app you'd need to deal with text/binary, and also consume streaming but here we know it will be strict)

johanandren · 2023-08-21T08:12:02Z

In general:

With a Thread.sleep you are effectively blocking the entire thread and Akka stream that handles the tcp socket, while .delay just simulates async-busy-work by letting the stream do other things.

Other would be collecting those TextMessages in a buffer for example. There is however no async boundary between source and delay, so there is no buffer introduced and .addAttributes(Attributes.inputBuffer(1, 1)) does not actually do anything.

Blocking a stream without async boundaries should also mean that the idle timeout operators are blocked from doing anything until that blocking stops, that interferes with what we want to test here.

gawronA · 2023-08-21T20:59:47Z

Ahh of course Flow.apply was the cause. I completely overlooked that one. Thank you for pointing it out :)

Other would be collecting those TextMessages in a buffer for example. There is however no async boundary between source and delay, so there is no buffer introduced and .addAttributes(Attributes.inputBuffer(1, 1)) does not actually do anything.

I would like to leave .addAttributes(Attributes.inputBuffer(1, 1)) as this causes the source to emit each element with a delay of 200ms. Removing it results in emitting all elements after the delay. The first case better reflects real world scenario.

johanandren · 2023-08-22T07:08:46Z

Right, delay has an internal buffer, so it is not about the async boundary buffers, I didn't remember that from the top of my head. Fine to keep.

johanandren · 2023-08-23T19:19:00Z

akka-http-core/src/test/scala/akka/http/impl/engine/ws/WebSocketServerSendIdleTimeoutSpec.scala

+      val myPort = binding.localAddress.getPort
+
+      Source(1 to 10).map(_ => {
+        Await.result(Future(Thread.sleep(200)), Duration(3, TimeUnit.SECONDS))


A final blocking sleep to get rid of, then I think this PR is good to go

* replace stream-blocking Await with .delay operator in websocket server timeout tests * add outgoing message stream from server in receive-idle-timeout test

* replace printing source with ignoring source in receive-idle-timeout server test

johanandren

LGTM, thanks, good work!

core: implement akka#2419

37a8df2

* Add receiveIdleTimeout and sendIdleTimeout config parameters to websocket client and server * Add idleTimeout stack layer * Add integration tests

johanandren reviewed Aug 14, 2023

View reviewed changes

core: apply review suggestions for akka#2419

2d6bd52

* Use BidiFlow.identity for default case instead of BidiFlow constructed from two identity Flows * Remove redundant maps * Add exception class matchers in tests

johanandren reviewed Aug 16, 2023

View reviewed changes

johanandren mentioned this pull request Aug 16, 2023

fix: Avoid turning all stream timeouts to TcpIdleTimeoutException akka/akka#32047

Open

core: add headers and apply minor PR suggestions akka#2419

d8b8998

johanandren reviewed Aug 21, 2023

View reviewed changes

core: replace stream-blocking await with .delay operator akka#2419

bd49dcb

core: fix formatting in tests akka#2419

8441d55

johanandren reviewed Aug 23, 2023

View reviewed changes

gawronA added 2 commits August 24, 2023 07:40

core: apply PR review akka#2419

47a831c

* replace stream-blocking Await with .delay operator in websocket server timeout tests * add outgoing message stream from server in receive-idle-timeout test

core: apply PR review akka#2419

e36f74c

* replace printing source with ignoring source in receive-idle-timeout server test

johanandren approved these changes Aug 24, 2023

View reviewed changes

johanandren merged commit 77c29a8 into akka:main Aug 24, 2023
16 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Websocket stream failure on receive/send timeout (network lost detection) #4289

Websocket stream failure on receive/send timeout (network lost detection) #4289

gawronA commented Jul 28, 2023

lightbend-cla-validator commented Jul 28, 2023

gawronA commented Jul 28, 2023

johanandren left a comment

gawronA commented Aug 15, 2023

johanandren left a comment

johanandren Aug 16, 2023

gawronA commented Aug 18, 2023 •

edited

Loading

johanandren Aug 21, 2023

johanandren commented Aug 21, 2023 •

edited

Loading

gawronA commented Aug 21, 2023

johanandren commented Aug 22, 2023

johanandren Aug 23, 2023

johanandren left a comment

Websocket stream failure on receive/send timeout (network lost detection) #4289

Websocket stream failure on receive/send timeout (network lost detection) #4289

Conversation

gawronA commented Jul 28, 2023

lightbend-cla-validator commented Jul 28, 2023

gawronA commented Jul 28, 2023

johanandren left a comment

Choose a reason for hiding this comment

gawronA commented Aug 15, 2023

johanandren left a comment

Choose a reason for hiding this comment

johanandren Aug 16, 2023

Choose a reason for hiding this comment

gawronA commented Aug 18, 2023 • edited Loading

johanandren Aug 21, 2023

Choose a reason for hiding this comment

johanandren commented Aug 21, 2023 • edited Loading

gawronA commented Aug 21, 2023

johanandren commented Aug 22, 2023

johanandren Aug 23, 2023

Choose a reason for hiding this comment

johanandren left a comment

Choose a reason for hiding this comment

gawronA commented Aug 18, 2023 •

edited

Loading

johanandren commented Aug 21, 2023 •

edited

Loading