Timeouts after 1000 ms

build.sbt

@@ -12,6 +12,7 @@ libraryDependencies ++= Seq(
   "com.typesafe.akka" %% "akka-http" % akkaHttpVersion,
   "com.typesafe.akka" %% "akka-http-spray-json" % akkaHttpVersion,
   "com.typesafe.akka" %% "akka-cluster-typed" % akkaVersion,
+  "com.typesafe.akka" %% "akka-actor" % akkaVersion,
   "io.spray" %% "spray-json" % "1.3.5",
   "ch.qos.logback" % "logback-classic" % "1.1.3" % Runtime,
   "org.scalactic" %% "scalactic" % "3.1.1",

src/main/scala/dynamodb/node/ExternalServer.scala

@@ -19,6 +19,7 @@ object ExternalServer {
   final case class Started(binding: ServerBinding) extends Command
   final case class StartFailed(cause: Throwable) extends Command
   final case class Stop() extends  Command
+  final case class Sleep(millis: Int) extends Command
 }
 
 class ExternalServer(context: ActorContext[ExternalServer.Command], valueRepository: ActorRef[ValueRepository.Command], internalClient: ActorRef[InternalClient.Command], host: String, port: Int)
@@ -55,6 +56,10 @@ class ExternalServer(context: ActorContext[ExternalServer.Command], valueReposit
 
         this
 
+      case Sleep(x) =>
+        Thread.sleep(x)
+        this
+
       case Stop() =>
         if (started) binding.unbind()
         context.log.info(

src/main/scala/dynamodb/node/InternalClient.scala

@@ -20,9 +20,10 @@ import dynamodb.node.ValueRepository.GetValueByKey
 
 import scala.collection.immutable.TreeMap
 import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Future
+import scala.concurrent.{Future, TimeoutException}
 import scala.concurrent.duration._
 
+
 object InternalClient {
   def apply(host: String, port: Int, numNodes: Int, numReadMinimum: Int, numWriteMinimum: Int, nodeName: String)(implicit valueRepository: ActorRef[ValueRepository.Command], dht: ActorRef[DistributedHashTable.Command]): Behavior[Command] = Behaviors.receive { (context, command) =>
     implicit val actorSystem: ActorSystem[Nothing] = context.system
@@ -86,8 +87,10 @@ object InternalClient {
   def getOtherNodes[T](key: String, address: Uri)(implicit actorSystem: actor.ActorSystem, mat: Materializer): Future[Option[ValueRepository.Value]] = {
     import JsonSupport._
 
+    val timeout = akka.pattern.after(Duration(1000, MILLISECONDS), using = actorSystem.scheduler) (Future.failed(new TimeoutException(s"timed out waiting on READ responses")))
+
     try for {
-      response <- Http().singleRequest(HttpRequest(uri = address + key))
+      response <- Future.firstCompletedOf(Seq(Http().singleRequest(HttpRequest(uri = address + key)), timeout))
       _ <- if (response.status != StatusCodes.OK) Future.failed(new Exception("Empty value")) else Future.successful()
       value <- Unmarshal(response).to[ValueRepository.Value]
     } yield Some(value) catch {
@@ -123,11 +126,16 @@ object InternalClient {
     import JsonSupport._
     import spray.json._
 
-    Http().singleRequest(HttpRequest(
-      method = HttpMethods.POST,
-      uri = address,
-      entity = HttpEntity(`application/json`, value.toJson.compactPrint)
-    ))
+    val timeout = akka.pattern.after(Duration(1000, MILLISECONDS), using = actorSystem.scheduler) (Future.failed(new TimeoutException(s"timed out waiting on WRITE responses")))
+
+    Future.firstCompletedOf( Seq(
+      Http().singleRequest(HttpRequest(
+        method = HttpMethods.POST,
+        uri = address,
+        entity = HttpEntity(`application/json`, value.toJson.compactPrint)
+      )),
+      timeout)
+    )
   }
 
   // Trait defining responses
@@ -145,4 +153,8 @@ object InternalClient {
 
   case object OK extends Response
 
+
+  def sleep(): Unit = {
+    Thread.sleep(5000)
+  }
 }

src/main/scala/dynamodb/node/InternalServer.scala

@@ -19,6 +19,7 @@ object InternalServer {
   final case class Started(binding: ServerBinding) extends Command
   final case class StartFailed(cause: Throwable) extends Command
   final case class Stop() extends Command
+  final case class Sleep(millis: Int) extends Command
 }
 
 class InternalServer(context: ActorContext[InternalServer.Command], valueRepository: ActorRef[ValueRepository.Command], host: String, port: Int)
@@ -55,6 +56,10 @@ class InternalServer(context: ActorContext[InternalServer.Command], valueReposit
 
         this
 
+      case Sleep(x) =>
+        Thread.sleep(x)
+        this
+
       case Stop() =>
         if (started) this.binding.unbind()
         context.log.info(

src/main/scala/dynamodb/node/Node.scala

@@ -14,6 +14,8 @@ object Node {
 
   sealed trait Message
 
+  final case class Sleep(millis: Int) extends Message
+
   private final case class StartFailed(cause: Throwable) extends Message
 
   private final case class Started(binding: ServerBinding) extends Message
@@ -26,6 +28,7 @@ object Node {
 
   case object Stop extends Message
 
+
   def apply(config: NodeConfig, allNodes: List[NodeConfig]): Behavior[Message] = Behaviors.setup { ctx =>
 
     implicit val scheduler: Scheduler = ctx.system.scheduler
@@ -50,6 +53,10 @@ object Node {
 
     def starting(): Behaviors.Receive[Message] =
       Behaviors.receiveMessagePartial[Message] {
+        case Sleep(x) =>
+          internalServer ! InternalServer.Sleep(x)
+          externalServer ! ExternalServer.Sleep(x)
+          Behaviors.same
         case Stop =>
           internalServer ! InternalServer.Stop()
           externalServer ! ExternalServer.Stop()

src/test/scala/dynamodb/node/ClusterSpec.scala

@@ -114,5 +114,22 @@ class ClusterSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll {
       post(coordinatorUrl, "/values", s"""{"key": "rejectedKey", "value": "myRejectedValue", "version": {"${coordinator}": 1}}""")
         .code shouldBe 400
     }
+
+    "fail read after top N nodes timeout " in {
+      val coordinator = getCoordinatorUrlForKey("timeKey")
+      val coordinatorUrl = hostToUrl(coordinator)
+
+      post(coordinatorUrl, "/values", """{"key": "timeKey", "value": "1"}""")
+        .body shouldBe "Value added"
+
+      // THis should succeed
+      get(coordinatorUrl, "/values/timeKey")
+        .body should be(s"""{"key":"timeKey","value":"1","version":{"${coordinator}":0}}""")
+
+      cluster(3) ! Node.Sleep(10000)
+      // This should fail
+      get(coordinatorUrl, "/values/timeKey")
+        .code shouldBe 200 // dit moet falen
+    }
   }
 }