Issues with Database Outage Recovery in Lagom 1.4.0

[octonato]

Up to now we believe there are two bugs involved:

• Slick DB not recovering after long period of failures
• RestartSource stopping to retry after long period of failures

[octonato]

After some further investigations, I came to the conclusion that there is nothing wrong with the RestartSource. It does restart forever.

All eyes are turned now to a possible deadlock in the Slick DB. Note that the Source that is fails to be restarted depends on two calls to the DB. One to fetch the Offset and another for the journal query.

RestartSource.withBackoff(
  config.minBackoff,
  config.maxBackoff,
  config.randomBackoffFactor
) { () =>
  val handler = processor().buildHandler
  val futureOffset = handler.prepare(tag) // <- this hits the DB
  Source.fromFuture(futureOffset).flatMapConcat {
    offset =>
      val eventStreamSource = eventStreamFactory(tag, offset)
      val userlandFlow = handler.handle()
      eventStreamSource.via(userlandFlow) // <- hits the DB again
  }
}

If the DB is somehow blocked, the Future with the Offset may never complete and that means that the Source wrapped by RestartSource will never fail because it never realy started. It will just hang forever in time and space.

[octonato]

Found the error in Slick. This is a regression that was introduced when fixing slick/slick#1274.

I have an open PR for Slick:
slick/slick#1876

[octonato]

In Lagom, we still need to fix the way we are using Source.fromFuture(futureOffset) (see code fragment above).

If futureOffset never completes, the Source will never start and the backoff will never restart it.

[TimMoore]

Maybe something like https://doc.akka.io/docs/akka/current/scala/futures.html#after or https://doc.akka.io/docs/akka/current/scala/stream/stages-overview.html#initialtimeout

[ignasi35]

@TimMoore if the problem is that handler.prepare(tag) doesn't time out (doesn't complete in some scenarios), then adding the timeout out from the outside would just hold more connections.

[octonato]

@TimMoore I fixed it locally using

   val eventualTimeout =
     after(duration = 3.seconds, using = context.system.scheduler) {
       Future.failed(new TimeoutException(s"Timeout while fetching read-side processor offset"))
     }
     val futureOffset = handler.prepare(tag).toScala
     val offsetWithTimeout = Future firstCompletedOf Seq(futureOffset, eventualTimeout)

@ignasi35, I rechecked the code and this is not true. The reason this particular future is never completing is that Slick AsyncExecutor doesn't have any available threads for it.

Whenever we get a Future from another library we can't make the assumption that it will complete so I think it's fair to say that one should always protect against it. Actually, we should bring that fix into akka streams. We should have a variant of Source.fromFuture that also receives a timeout and Source should fail if it never completes. I will raise an issue there to know whay they think.

[octonato]

Oh, it seems that there is already something we can use.

Source.fromFuture(fut).completionTimeout(3.seconds)

[octonato]

There is an open PR (#1278) that fix the RestartSource issue. I'm keeping this open to track the fix in Slick.