diff --git a/build.sbt b/build.sbt index 24056e0..fc848f5 100644 --- a/build.sbt +++ b/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", diff --git a/src/main/scala/dynamodb/node/ExternalServer.scala b/src/main/scala/dynamodb/node/ExternalServer.scala index a325cc0..268d050 100644 --- a/src/main/scala/dynamodb/node/ExternalServer.scala +++ b/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( diff --git a/src/main/scala/dynamodb/node/InternalClient.scala b/src/main/scala/dynamodb/node/InternalClient.scala index aa90381..d48151e 100644 --- a/src/main/scala/dynamodb/node/InternalClient.scala +++ b/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) + } } diff --git a/src/main/scala/dynamodb/node/InternalServer.scala b/src/main/scala/dynamodb/node/InternalServer.scala index cdfd625..3e75cbf 100644 --- a/src/main/scala/dynamodb/node/InternalServer.scala +++ b/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( diff --git a/src/main/scala/dynamodb/node/Node.scala b/src/main/scala/dynamodb/node/Node.scala index 9b06d80..97f0228 100644 --- a/src/main/scala/dynamodb/node/Node.scala +++ b/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() diff --git a/src/test/scala/dynamodb/node/ClusterSpec.scala b/src/test/scala/dynamodb/node/ClusterSpec.scala index 1a590f5..cc214dc 100644 --- a/src/test/scala/dynamodb/node/ClusterSpec.scala +++ b/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 + } } }