Provide better diagnostics when handlers crash (#4091)

In the event when the action of the underlying actor crashes really bad, the result of the future will be `Failure`. All such results will be then wrapped in `Status.Failure` via `pipeTo` (unlike `Success` which just forwards the response).

In some cases we don't handle `Status.Failure` messages, meaning we are rather left in the dark to the reason of the failure with a non-informative message `Received unknown message: class akka.actor.Status$Failure`.

I'm keeping this change small on purpose to keep this change self-contained. I think there are more cases but it needs careful investigation into how messages are being sent.

# Important Notes
This PR does not attempt to fix the underlying problem of the ticket, yet. We should however have a better overview where/why things go wrong.

engine/language-server/src/main/scala/org/enso/languageserver/requesthandler/vcs/InitVcsHandler.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.requesthandler.vcs
 
-import akka.actor.{Actor, ActorRef, Cancellable, Props}
+import akka.actor.{Actor, ActorRef, Cancellable, Props, Status}
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.jsonrpc.{
   Errors,
@@ -52,6 +52,15 @@ class InitVcsHandler(
       replyTo ! ResponseError(Some(id), Errors.RequestTimeout)
       context.stop(self)
 
+    case Status.Failure(ex) =>
+      logger.error(
+        s"Initialize project request [$id] for [${rpcSession.clientId}] failed with: ${ex.getMessage}.",
+        ex
+      )
+      cancellable.cancel()
+      replyTo ! ResponseError(Some(id), Errors.ServiceError)
+      context.stop(self)
+
     case VcsProtocol.InitRepoResponse(Right(_)) =>
       replyTo ! ResponseResult(InitVcs, id, Unused)
       cancellable.cancel()

engine/language-server/src/main/scala/org/enso/languageserver/requesthandler/vcs/ListVcsHandler.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.requesthandler.vcs
 
-import akka.actor.{Actor, ActorRef, Cancellable, Props}
+import akka.actor.{Actor, ActorRef, Cancellable, Props, Status}
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.jsonrpc._
 import org.enso.languageserver.requesthandler.RequestTimeout
@@ -49,6 +49,15 @@ class ListVcsHandler(
       replyTo ! ResponseError(Some(id), Errors.RequestTimeout)
       context.stop(self)
 
+    case Status.Failure(ex) =>
+      logger.error(
+        s"List project request [$id] for [${rpcSession.clientId}] failed with: ${ex.getMessage}.",
+        ex
+      )
+      cancellable.cancel()
+      replyTo ! ResponseError(Some(id), Errors.ServiceError)
+      context.stop(self)
+
     case VcsProtocol.ListRepoResponse(Right(saves)) =>
       replyTo ! ResponseResult(
         ListVcs,

engine/language-server/src/main/scala/org/enso/languageserver/requesthandler/vcs/RestoreVcsHandler.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.requesthandler.vcs
 
-import akka.actor.{Actor, ActorRef, Cancellable, Props}
+import akka.actor.{Actor, ActorRef, Cancellable, Props, Status}
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.jsonrpc._
 import org.enso.languageserver.requesthandler.RequestTimeout
@@ -51,6 +51,15 @@ class RestoreVcsHandler(
       replyTo ! ResponseError(Some(id), Errors.RequestTimeout)
       context.stop(self)
 
+    case Status.Failure(ex) =>
+      logger.error(
+        s"Restore project request [$id] for [${rpcSession.clientId}] failed with: ${ex.getMessage}.",
+        ex
+      )
+      cancellable.cancel()
+      replyTo ! ResponseError(Some(id), Errors.ServiceError)
+      context.stop(self)
+
     case VcsProtocol.RestoreRepoResponse(Right(paths)) =>
       replyTo ! ResponseResult(RestoreVcs, id, RestoreVcs.Result(paths))
       cancellable.cancel()

engine/language-server/src/main/scala/org/enso/languageserver/requesthandler/vcs/SaveVcsHandler.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.requesthandler.vcs
 
-import akka.actor.{Actor, ActorRef, Cancellable, Props}
+import akka.actor.{Actor, ActorRef, Cancellable, Props, Status}
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.jsonrpc.{Errors, Id, Request, ResponseError, ResponseResult}
 import org.enso.languageserver.requesthandler.RequestTimeout
@@ -51,6 +51,15 @@ class SaveVcsHandler(
       replyTo ! ResponseError(Some(id), Errors.RequestTimeout)
       context.stop(self)
 
+    case Status.Failure(ex) =>
+      logger.error(
+        s"Save project request [$id] for [${rpcSession.clientId}] failed with: ${ex.getMessage}.",
+        ex
+      )
+      cancellable.cancel()
+      replyTo ! ResponseError(Some(id), Errors.ServiceError)
+      context.stop(self)
+
     case VcsProtocol.SaveRepoResponse(Right((name, sha))) =>
       replyTo ! ResponseResult(SaveVcs, id, SaveVcs.Result(name, sha))
       cancellable.cancel()

engine/language-server/src/main/scala/org/enso/languageserver/requesthandler/vcs/StatusVcsHandler.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.requesthandler.vcs
 
-import akka.actor.{Actor, ActorRef, Cancellable, Props}
+import akka.actor.{Actor, ActorRef, Cancellable, Props, Status}
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.jsonrpc.{Errors, Id, Request, ResponseError, ResponseResult}
 import org.enso.languageserver.requesthandler.RequestTimeout
@@ -45,6 +45,15 @@ class StatusVcsHandler(
       replyTo ! ResponseError(Some(id), Errors.RequestTimeout)
       context.stop(self)
 
+    case Status.Failure(ex) =>
+      logger.error(
+        s"Status project request [$id] for [${rpcSession.clientId}] failed with: ${ex.getMessage}.",
+        ex
+      )
+      cancellable.cancel()
+      replyTo ! ResponseError(Some(id), Errors.ServiceError)
+      context.stop(self)
+
     case VcsProtocol.StatusRepoResponse(Right((isModified, changed, last))) =>
       replyTo ! ResponseResult(
         StatusVcs,

engine/language-server/src/main/scala/org/enso/languageserver/text/CollaborativeBuffer.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.text
 
-import akka.actor.{Actor, ActorRef, Cancellable, Props, Stash}
+import akka.actor.{Actor, ActorRef, Cancellable, Props, Stash, Status}
 import cats.implicits._
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.languageserver.boot.TimingsConfig
@@ -11,15 +11,11 @@ import org.enso.languageserver.event.{
   BufferOpened,
   JsonSessionTerminated
 }
-import org.enso.languageserver.filemanager.FileManagerProtocol.{
-  ReadTextualFileResult,
-  TextualFileContent,
-  WriteFileResult
-}
 import org.enso.languageserver.filemanager.{
   FileEventKind,
   FileManagerProtocol,
   FileNotFound,
+  GenericFileSystemFailure,
   OperationTimeout,
   Path
 }
@@ -95,12 +91,12 @@ class CollaborativeBuffer(
     timeoutCancellable: Cancellable,
     inMemoryBuffer: Boolean
   ): Receive = {
-    case ReadTextualFileResult(Right(content)) =>
+    case FileManagerProtocol.ReadTextualFileResult(Right(content)) =>
       handleFileContent(rpcSession, replyTo, content, inMemoryBuffer, Map.empty)
       unstashAll()
       timeoutCancellable.cancel()
 
-    case ReadTextualFileResult(Left(failure)) =>
+    case FileManagerProtocol.ReadTextualFileResult(Left(failure)) =>
       replyTo ! OpenFileResponse(Left(failure))
       timeoutCancellable.cancel()
       stop(Map.empty)
@@ -109,6 +105,17 @@ class CollaborativeBuffer(
       replyTo ! OpenFileResponse(Left(OperationTimeout))
       stop(Map.empty)
 
+    case Status.Failure(failure) =>
+      logger.error(
+        s"Waiting for file content for [${rpcSession.clientId}] failed with: ${failure.getMessage}.",
+        failure
+      )
+      replyTo ! OpenFileResponse(
+        Left(GenericFileSystemFailure(failure.getMessage))
+      )
+      timeoutCancellable.cancel()
+      stop(Map.empty)
+
     case _ => stash()
   }
 
@@ -253,7 +260,7 @@ class CollaborativeBuffer(
     clients: Map[ClientId, JsonSession],
     inMemoryBuffer: Boolean
   ): Receive = {
-    case ReadTextualFileResult(Right(file)) =>
+    case FileManagerProtocol.ReadTextualFileResult(Right(file)) =>
       timeoutCancellable.cancel()
       val buffer = Buffer(file.path, file.content, inMemoryBuffer)
 
@@ -281,7 +288,7 @@ class CollaborativeBuffer(
         )
       )
 
-    case ReadTextualFileResult(Left(FileNotFound)) =>
+    case FileManagerProtocol.ReadTextualFileResult(Left(FileNotFound)) =>
       clients.values.foreach {
         _.rpcController ! TextProtocol.FileEvent(path, FileEventKind.Removed)
       }
@@ -289,7 +296,7 @@ class CollaborativeBuffer(
       timeoutCancellable.cancel()
       stop(Map.empty)
 
-    case ReadTextualFileResult(Left(err)) =>
+    case FileManagerProtocol.ReadTextualFileResult(Left(err)) =>
       replyTo ! ReloadBufferFailed(path, "io failure: " + err.toString)
       timeoutCancellable.cancel()
       context.become(
@@ -301,6 +308,15 @@ class CollaborativeBuffer(
         )
       )
 
+    case Status.Failure(ex) =>
+      logger.error(
+        s"Waiting for file content for [${rpcSession.clientId}] failed with: ${ex.getMessage}.",
+        ex
+      )
+      replyTo ! ReloadBufferFailed(path, "io failure: " + ex.toString)
+      timeoutCancellable.cancel()
+      stop(Map.empty)
+
     case IOTimeout =>
       replyTo ! ReloadBufferFailed(path, "io timeout")
       context.become(
@@ -337,7 +353,7 @@ class CollaborativeBuffer(
           )
       }
 
-    case WriteFileResult(Left(failure)) =>
+    case FileManagerProtocol.WriteFileResult(Left(failure)) =>
       replyTo.foreach(_ ! SaveFailed(failure))
       unstashAll()
       timeoutCancellable.cancel()
@@ -351,7 +367,23 @@ class CollaborativeBuffer(
           )
       }
 
-    case WriteFileResult(Right(())) =>
+    case Status.Failure(failure) =>
+      logger.error(
+        s"Waiting on save operation to complete failed with: ${failure.getMessage}.",
+        failure
+      )
+      timeoutCancellable.cancel()
+      onClose match {
+        case Some(clientId) =>
+          replyTo.foreach(_ ! FileClosed)
+          removeClient(buffer, clients, lockHolder, clientId, autoSave)
+        case None =>
+          context.become(
+            collaborativeEditing(buffer, clients, lockHolder, autoSave)
+          )
+      }
+
+    case FileManagerProtocol.WriteFileResult(Right(())) =>
       replyTo match {
         case Some(replyTo) => replyTo ! FileSaved
         case None =>
@@ -624,7 +656,7 @@ class CollaborativeBuffer(
   private def handleFileContent(
     rpcSession: JsonSession,
     originalSender: ActorRef,
-    file: TextualFileContent,
+    file: FileManagerProtocol.TextualFileContent,
     inMemoryBuffer: Boolean,
     autoSave: Map[ClientId, (ContentVersion, Cancellable)]
   ): Unit = {