fix(triplestore-connector): stack crashes on invalid search (DEV-1154)

dsp-shared/src/main/scala/dsp/errors/Errors.scala

@@ -106,6 +106,9 @@ case class ForbiddenException(message: String) extends RequestRejectedException(
  * @param message a description of the error.
  */
 case class NotFoundException(message: String) extends RequestRejectedException(message)
+object NotFoundException {
+  val notFound = NotFoundException("The requested data was not found")
+}
 
 /**
  * An exception indicating that a requested update is not allowed because it would create a duplicate value.
@@ -271,91 +274,6 @@ object AssertionException {
     AssertionException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
 }
 
-/**
- * An abstract class for exceptions indicating that something went wrong with the triplestore.
- *
- * @param message a description of the error.
- * @param cause   the original exception representing the cause of the error, if any.
- */
-abstract class TriplestoreException(message: String, cause: Option[Throwable] = None)
-    extends InternalServerException(message, cause)
-
-/**
- * Indicates that the network connection to the triplestore failed.
- *
- * @param message a description of the error.
- * @param cause   the original exception representing the cause of the error, if any.
- */
-case class TriplestoreConnectionException(message: String, cause: Option[Throwable] = None)
-    extends TriplestoreException(message, cause)
-
-object TriplestoreConnectionException {
-  def apply(message: String, e: Throwable, log: Logger): TriplestoreConnectionException =
-    TriplestoreConnectionException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
-}
-
-/**
- * Indicates that a read timeout occurred while waiting for data from the triplestore.
- *
- * @param message a description of the error.
- * @param cause   the original exception representing the cause of the error, if any.
- */
-final case class TriplestoreTimeoutException(message: String, cause: Option[Throwable] = None)
-    extends TriplestoreException(message, cause)
-
-object TriplestoreTimeoutException {
-  def apply(message: String, e: Throwable, log: Logger): TriplestoreTimeoutException =
-    TriplestoreTimeoutException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
-
-  def apply(message: String, cause: Throwable): TriplestoreTimeoutException =
-    TriplestoreTimeoutException(message, Some(cause))
-}
-
-/**
- * Indicates that we tried using a feature which is unsuported by the selected triplestore.
- *
- * @param message a description of the error.
- * @param cause   the original exception representing the cause of the error, if any.
- */
-case class TriplestoreUnsupportedFeatureException(message: String, cause: Option[Throwable] = None)
-    extends TriplestoreException(message, cause)
-
-object TriplestoreUnsupportedFeatureException {
-  def apply(message: String, e: Throwable, log: Logger): TriplestoreUnsupportedFeatureException =
-    TriplestoreUnsupportedFeatureException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
-}
-
-/**
- * Indicates that something inside the Triplestore package went wrong. More details can be given in the message parameter.
- *
- * @param message a description of the error.
- * @param cause   the original exception representing the cause of the error, if any.
- */
-case class TriplestoreInternalException(message: String, cause: Option[Throwable] = None)
-    extends TriplestoreException(message, cause)
-
-object TriplestoreInternalException {
-  def apply(message: String, e: Throwable, log: Logger): TriplestoreInternalException =
-    TriplestoreInternalException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
-}
-
-/**
- * Indicates that the triplestore returned an error message, or a response that could not be parsed.
- *
- * @param message a description of the error.
- * @param cause   the original exception representing the cause of the error, if any.
- */
-final case class TriplestoreResponseException(message: String, cause: Option[Throwable] = None)
-    extends TriplestoreException(message, cause)
-
-object TriplestoreResponseException {
-  def apply(message: String, e: Throwable, log: Logger): TriplestoreResponseException =
-    TriplestoreResponseException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
-
-  def apply(message: String): TriplestoreResponseException =
-    TriplestoreResponseException(message)
-}
-
 /**
  * Indicates an inconsistency in repository data.
  *
@@ -452,22 +370,6 @@ case class FileWriteException(message: String) extends InternalServerException(m
  */
 case class NotImplementedException(message: String) extends InternalServerException(message)
 
-/**
- * Indicates that an error occurred with Sipi not relating to the user's request (it is not the user's fault).
- *
- * @param message a description of the error.
- */
-case class SipiException(message: String, cause: Option[Throwable] = None)
-    extends InternalServerException(message, cause)
-
-object SipiException {
-  def apply(message: String, e: Throwable, log: Logger): SipiException =
-    SipiException(message, Some(ExceptionUtil.logAndWrapIfNotSerializable(e, log)))
-
-  def apply(message: String, e: Throwable): SipiException =
-    SipiException(message, Some(e))
-}
-
 /**
  * An abstract base class for exceptions indicating that something about a configuration made it impossible to start.
  *

webapi/src/main/resources/application.conf

@@ -149,39 +149,6 @@ akka {
     }
 }
 
-akka.actor.deployment {
-
-    /applicationManager/storeManager/triplestoreManager/httpTriplestoreRouter {
-        router = balancing-pool
-        nr-of-instances = 10
-        nr-of-instances = ${?KNORA_WEBAPI_DB_CONNECTIONS}
-        pool-dispatcher {
-            executor = "thread-pool-executor"
-
-            # allocate exactly 10 threads for this pool
-            thread-pool-executor {
-                core-pool-size-min = 2
-                core-pool-size-min = ${?KNORA_WEBAPI_DB_CONNECTIONS}
-                core-pool-size-max = 2
-                core-pool-size-max = ${?KNORA_WEBAPI_DB_CONNECTIONS}
-            }
-        }
-    }
-
-    /applicationManager/storeManager/iiifManager/sipiConnector {
-        router = balancing-pool
-        nr-of-instances = 10
-        pool-dispatcher {
-            executor = "thread-pool-executor"
-
-            # allocate exactly 10 threads for this pool
-            thread-pool-executor {
-                core-pool-size-min = 10
-                core-pool-size-max = 10
-            }
-        }
-    }
-}
 
 // all responder actors should run on this dispatcher
 knora-actor-dispatcher {

webapi/src/main/scala/org/knora/webapi/app/ApplicationActor.scala

@@ -27,7 +27,6 @@ import org.knora.webapi.config.AppConfig
 import org.knora.webapi.core.LiveActorMaker
 import dsp.errors.InconsistentRepositoryDataException
 import dsp.errors.MissingLastModificationDateOntologyException
-import dsp.errors.SipiException
 import dsp.errors.UnexpectedMessageException
 import dsp.errors.UnsupportedValueException
 import org.knora.webapi.http.directives.DSPApiDirectives
@@ -58,6 +57,7 @@ import org.knora.webapi.settings._
 import org.knora.webapi.store.cache.CacheServiceManager
 import org.knora.webapi.store.cache.settings.CacheServiceSettings
 import org.knora.webapi.store.iiif.IIIFServiceManager
+import org.knora.webapi.store.iiif.errors.SipiException
 import org.knora.webapi.util.ActorUtil.future2Message
 import org.knora.webapi.util.cache.CacheUtil
 import redis.clients.jedis.exceptions.JedisConnectionException
@@ -71,6 +71,8 @@ import org.knora.webapi.messages.store.cacheservicemessages.CacheServiceRequest
 import org.knora.webapi.messages.store.sipimessages.IIIFRequest
 import org.knora.webapi.util.ActorUtil
 import org.knora.webapi.store.triplestore.TriplestoreServiceManager
+import org.knora.webapi.messages.ResponderRequest
+import akka.routing.RoundRobinPool
 
 /**
  * This is the first actor in the application. All other actors are children
@@ -138,6 +140,22 @@ class ApplicationActor(
     appActor = self
   )
 
+  val routerActor =
+    context.actorOf(
+      RoundRobinPool(1000).props(
+        Props(
+          new ApplicationRouterActor(
+            responderManager,
+            cacheServiceManager,
+            iiifServiceManager,
+            triplestoreManager,
+            appConfig
+          )
+        ).withDispatcher(KnoraDispatchers.KnoraActorDispatcher)
+      ),
+      "RouterActor"
+    )
+
   /**
    * This actor acts as the supervisor for its child actors.
    * Here we can override the default supervisor strategy.
@@ -194,6 +212,7 @@ class ApplicationActor(
   }
 
   def ready(): Receive = {
+
     /* Usually only called from tests */
     case AppStop() =>
       appStop()
@@ -373,12 +392,10 @@ class ApplicationActor(
       timers.startSingleTimer("CheckCacheService", CheckCacheService, 5.seconds)
 
     // Forward messages to the responder manager and the different store managers.
-    case msg: KnoraRequestV1      => future2Message(sender(), responderManager.receive(msg), log)
-    case msg: KnoraRequestV2      => future2Message(sender(), responderManager.receive(msg), log)
-    case msg: KnoraRequestADM     => future2Message(sender(), responderManager.receive(msg), log)
-    case msg: CacheServiceRequest => ActorUtil.zio2Message(sender(), cacheServiceManager.receive(msg), appConfig)
-    case msg: IIIFRequest         => ActorUtil.zio2Message(sender(), iiifServiceManager.receive(msg), appConfig)
-    case msg: TriplestoreRequest  => ActorUtil.zio2Message(sender(), triplestoreManager.receive(msg), appConfig)
+    case msg: ResponderRequest    => routerActor.forward(msg)
+    case msg: CacheServiceRequest => routerActor.forward(msg)
+    case msg: IIIFRequest         => routerActor.forward(msg)
+    case msg: TriplestoreRequest  => routerActor.forward(msg)
 
     case akka.actor.Status.Failure(ex: Exception) =>
       ex match {

webapi/src/main/scala/org/knora/webapi/app/ApplicationRouterActor.scala

@@ -0,0 +1,34 @@
+package org.knora.webapi.app
+
+import akka.actor.Actor
+import org.knora.webapi.messages.ResponderRequest
+import com.typesafe.scalalogging.Logger
+import org.knora.webapi.responders.ResponderManager
+import org.knora.webapi.util.ActorUtil
+import scala.concurrent.ExecutionContext
+import org.knora.webapi.messages.store.cacheservicemessages.CacheServiceRequest
+import org.knora.webapi.messages.store.sipimessages.IIIFRequest
+import org.knora.webapi.messages.store.triplestoremessages.TriplestoreRequest
+import org.knora.webapi.store.iiif.IIIFServiceManager
+import org.knora.webapi.store.triplestore.TriplestoreServiceManager
+import org.knora.webapi.store.cache.CacheServiceManager
+import org.knora.webapi.config.AppConfig
+
+class ApplicationRouterActor(
+  responderManager: ResponderManager,
+  cacheServiceManager: CacheServiceManager,
+  iiifServiceManager: IIIFServiceManager,
+  triplestoreManager: TriplestoreServiceManager,
+  appConfig: AppConfig
+) extends Actor {
+  val log: Logger                   = Logger(this.getClass())
+  implicit val ec: ExecutionContext = context.dispatcher
+  def receive: Receive = {
+    case msg: ResponderRequest.KnoraRequestV1  => ActorUtil.future2Message(sender(), responderManager.receive(msg), log)
+    case msg: ResponderRequest.KnoraRequestV2  => ActorUtil.future2Message(sender(), responderManager.receive(msg), log)
+    case msg: ResponderRequest.KnoraRequestADM => ActorUtil.future2Message(sender(), responderManager.receive(msg), log)
+    case msg: CacheServiceRequest              => ActorUtil.zio2Message(sender(), cacheServiceManager.receive(msg), appConfig, log)
+    case msg: IIIFRequest                      => ActorUtil.zio2Message(sender(), iiifServiceManager.receive(msg), appConfig, log)
+    case msg: TriplestoreRequest               => ActorUtil.zio2Message(sender(), triplestoreManager.receive(msg), appConfig, log)
+  }
+}

webapi/src/main/scala/org/knora/webapi/http/status/ApiStatusCodesV2.scala

@@ -8,6 +8,7 @@ package org.knora.webapi.http.status
 import akka.http.scaladsl.model.StatusCode
 import akka.http.scaladsl.model.StatusCodes
 import dsp.errors._
+import org.knora.webapi.store.triplestore.errors.TriplestoreTimeoutException
 
 /**
  * The possible values for the HTTP status code that is returned as part of each Knora API v2 response.

webapi/src/main/scala/org/knora/webapi/messages/v2/responder/valuemessages/ValueMessagesV2.scala

@@ -13,7 +13,6 @@ import com.typesafe.scalalogging.Logger
 import dsp.errors.AssertionException
 import dsp.errors.BadRequestException
 import dsp.errors.NotImplementedException
-import dsp.errors.SipiException
 import dsp.valueobjects.IriErrorMessages
 import org.knora.webapi._
 import org.knora.webapi.messages.IriConversions._
@@ -36,6 +35,7 @@ import org.knora.webapi.messages.v2.responder.resourcemessages.ReadResourceV2
 import org.knora.webapi.messages.v2.responder.standoffmessages._
 import org.knora.webapi.responders.ResponderManager
 import org.knora.webapi.settings.KnoraSettingsImpl
+import org.knora.webapi.store.iiif.errors.SipiException
 import org.knora.webapi.util._
 
 import java.time.Instant

webapi/src/main/scala/org/knora/webapi/responders/v1/StandoffResponderV1.scala

@@ -8,7 +8,6 @@ package org.knora.webapi.responders.v1
 import akka.pattern._
 import org.knora.webapi._
 import dsp.errors.NotFoundException
-import dsp.errors.SipiException
 
 import org.knora.webapi.messages.IriConversions._
 import org.knora.webapi.messages.StringFormatter
@@ -20,6 +19,7 @@ import org.knora.webapi.messages.v1.responder.standoffmessages._
 import org.knora.webapi.messages.v2.responder.standoffmessages._
 import org.knora.webapi.responders.Responder
 import org.knora.webapi.responders.Responder.handleUnexpectedMessage
+import org.knora.webapi.store.iiif.errors.SipiException
 
 import java.util.UUID
 import scala.concurrent.Future

webapi/src/main/scala/org/knora/webapi/responders/v2/ResourceUtilV2.scala

@@ -215,7 +215,10 @@ object ResourceUtilV2 {
             )
 
             // If Sipi succeeds, return the future we were given. Otherwise, return a failed future.
-            appActor.ask(sipiRequest).mapTo[SuccessResponseV2].flatMap(_ => updateFuture)
+            appActor
+              .ask(sipiRequest)
+              .mapTo[SuccessResponseV2]
+              .flatMap(_ => updateFuture)
 
           case Failure(_) =>
             // The file value update failed. Ask Sipi to delete the temporary file.
@@ -224,11 +227,14 @@ object ResourceUtilV2 {
               requestingUser = requestingUser
             )
 
-            val sipiResponseFuture: Future[SuccessResponseV2] = appActor.ask(sipiRequest).mapTo[SuccessResponseV2]
+            val sipiResponseFuture: Future[SuccessResponseV2] =
+              appActor
+                .ask(sipiRequest)
+                .mapTo[SuccessResponseV2]
 
             // Did Sipi successfully delete the temporary file?
             sipiResponseFuture.transformWith {
-              case Success(_) =>
+              case Success(value) =>
                 // Yes. Return the future we were given.
                 updateFuture
 

webapi/src/main/scala/org/knora/webapi/responders/v2/ResourcesResponderV2.scala

@@ -52,6 +52,7 @@ import org.knora.webapi.messages.v2.responder.standoffmessages.GetXSLTransformat
 import org.knora.webapi.messages.v2.responder.valuemessages._
 import org.knora.webapi.responders.IriLocker
 import org.knora.webapi.responders.Responder.handleUnexpectedMessage
+import org.knora.webapi.store.iiif.errors.SipiException
 import org.knora.webapi.util._
 
 import java.time.Instant

webapi/src/main/scala/org/knora/webapi/responders/v2/ResponderWithStandoffV2.scala

@@ -9,7 +9,6 @@ import akka.http.scaladsl.util.FastFuture
 import akka.pattern._
 import org.knora.webapi.IRI
 import dsp.errors.NotFoundException
-import dsp.errors.SipiException
 
 import org.knora.webapi.messages.admin.responder.usersmessages.UserADM
 import org.knora.webapi.messages.util.ConstructResponseUtilV2
@@ -21,6 +20,7 @@ import org.knora.webapi.messages.v2.responder.standoffmessages.GetMappingRespons
 import org.knora.webapi.messages.v2.responder.standoffmessages.GetXSLTransformationRequestV2
 import org.knora.webapi.messages.v2.responder.standoffmessages.GetXSLTransformationResponseV2
 import org.knora.webapi.responders.Responder
+import org.knora.webapi.store.iiif.errors.SipiException
 
 import scala.concurrent.Future
 

webapi/src/main/scala/org/knora/webapi/responders/v2/SearchResponderV2.scala

@@ -12,7 +12,6 @@ import dsp.errors.AssertionException
 import dsp.errors.BadRequestException
 import dsp.errors.GravsearchException
 import dsp.errors.InconsistentRepositoryDataException
-import dsp.errors.TriplestoreTimeoutException
 
 import org.knora.webapi.messages.IriConversions._
 import org.knora.webapi.messages.OntologyConstants
@@ -42,6 +41,7 @@ import org.knora.webapi.messages.v2.responder.ontologymessages.ReadPropertyInfoV
 import org.knora.webapi.messages.v2.responder.resourcemessages._
 import org.knora.webapi.messages.v2.responder.searchmessages._
 import org.knora.webapi.responders.Responder.handleUnexpectedMessage
+import org.knora.webapi.store.triplestore.errors.TriplestoreTimeoutException
 import org.knora.webapi.util.ApacheLuceneSupport._
 
 import scala.concurrent.Future