89 queue buffer overflow

.gitignore

@@ -3,5 +3,8 @@ target/
 
 # IDE
 .idea
+.metals/
+project/.bloop
+.bloop
 
 .pairs

.scalafmt.conf

@@ -1,3 +1,5 @@
+version = "2.0.1"
+
 style = defaultWithAlign
 
 align = most

scheduler/src/main/scala/com/sky/kms/actors/PublisherActor.scala

@@ -2,19 +2,21 @@ package com.sky.kms.actors
 
 import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
 import akka.stream.QueueOfferResult
+import akka.pattern.pipe
 import akka.stream.scaladsl.SourceQueueWithComplete
 import cats.syntax.show._
 import com.sky.kms.Start
-import com.sky.kms.actors.PublisherActor.{DownstreamFailure, Init, ScheduleQueue, Trigger}
+import com.sky.kms.actors.PublisherActor.{BufferUnblocked, DownstreamFailure, Init, ScheduleQueue, Trigger}
 import com.sky.kms.domain.PublishableMessage.ScheduledMessage
 import com.sky.kms.domain.{ScheduleEvent, ScheduleId, ScheduleQueueOfferResult}
 
 import scala.util.{Failure, Success}
+import scala.concurrent.Future
+import akka.actor.Stash
 
-class PublisherActor extends Actor with ActorLogging {
-
-  implicit val ec = context.dispatcher
+class PublisherActor extends Actor with ActorLogging with Stash {
 
+  implicit val ec               = context.dispatcher
   override def receive: Receive = waitForQueue
 
   private def waitForQueue: Receive = {
@@ -23,17 +25,30 @@ class PublisherActor extends Actor with ActorLogging {
       context become (receiveWithQueue(queue) orElse stopOnError)
   }
 
+  private def waitingState(queue: ScheduleQueue): Receive = stopOnError orElse {
+    case BufferUnblocked =>
+      context become (receiveWithQueue(queue) orElse stopOnError)
+      unstashAll()
+    case _ => stash()
+  }
+
   private def receiveWithQueue(queue: ScheduleQueue): Receive = {
     case Trigger(scheduleId, schedule) =>
-      queue.offer((scheduleId, messageFrom(schedule))) onComplete {
-        case Success(QueueOfferResult.Enqueued) =>
-          log.debug(ScheduleQueueOfferResult(scheduleId, QueueOfferResult.Enqueued).show)
-        case Success(res) =>
-          log.warning(ScheduleQueueOfferResult(scheduleId, res).show)
-        case Failure(t) =>
-          log.error(t, s"Failed to enqueue $scheduleId")
-          self ! DownstreamFailure(t)
-      }
+      context become waitingState(queue)
+      queue
+        .offer((scheduleId, messageFrom(schedule)))
+        .transformWith {
+          case Success(QueueOfferResult.Enqueued) =>
+            log.debug(ScheduleQueueOfferResult(scheduleId, QueueOfferResult.Enqueued).show)
+            Future.successful(BufferUnblocked)
+          case Success(res) =>
+            log.warning(ScheduleQueueOfferResult(scheduleId, res).show)
+            Future.successful(BufferUnblocked)
+          case Failure(t) =>
+            log.error(t, s"Failed to enqueue $scheduleId")
+            self ! DownstreamFailure(t)
+            Future.successful(BufferUnblocked)
+        } pipeTo self
   }
 
   private def stopOnError: Receive = {
@@ -54,6 +69,8 @@ object PublisherActor {
 
   case class Trigger(scheduleId: ScheduleId, schedule: ScheduleEvent)
 
+  case object BufferUnblocked
+
   case class DownstreamFailure(t: Throwable)
 
   def create(implicit system: ActorSystem): ActorRef =

scheduler/src/test/scala/com/sky/kms/unit/PublisherActorSpec.scala

@@ -10,20 +10,29 @@ import com.sky.kms.utils.TestDataUtils._
 import com.sky.kms.domain.{ScheduleEvent, ScheduleId}
 import org.mockito.Mockito._
 import org.scalatestplus.mockito.MockitoSugar
+import akka.stream.scaladsl.Source
+import akka.stream.OverflowStrategy
+import akka.stream.scaladsl.Sink
+import akka.stream.scaladsl.Keep
+import akka.stream.ActorMaterializer
+import com.sky.kms.domain.PublishableMessage.ScheduledMessage
 
 import scala.concurrent.Future
+import scala.concurrent.duration._
 
 class PublisherActorSpec extends AkkaSpecBase with MockitoSugar {
 
   "A publisher actor" must {
 
     "add scheduled messages to the queue" in new TestContext {
+      initQueue
       val (scheduleId, schedule) = generateSchedule
       publisherActor ! Trigger(scheduleId, schedule)
       verify(mockSourceQueue).offer((scheduleId, schedule.toScheduledMessage))
     }
 
     "stop when offering to the queue fails because the buffer is full" in new TestContext {
+      initQueue
       watch(publisherActor)
       val (scheduleId, schedule) = generateSchedule
       when(mockSourceQueue.offer((scheduleId, schedule.toScheduledMessage)))
@@ -34,7 +43,27 @@ class PublisherActorSpec extends AkkaSpecBase with MockitoSugar {
       expectTerminated(publisherActor)
     }
 
+    "do not offer when queue backpressures" in new TestContext {
+
+      val (_, schedule) = generateSchedule
+
+      implicit val mat = ActorMaterializer()
+
+      watch(publisherActor)
+      val (sourceQueue, _) = Source
+        .queue[(String, ScheduledMessage)](1, OverflowStrategy.backpressure)
+        .throttle(50, 1.second)
+        .toMat(Sink.queue())(Keep.both)
+        .run()
+
+      publisherActor ! Init(sourceQueue)
+      (1 to 1500).map { id =>
+        publisherActor ! Trigger(id.toString, schedule)
+      }
+    }
+
     "stop when offering to the queue fails" in new TestContext {
+      initQueue
       watch(publisherActor)
       val (scheduleId, schedule) = generateSchedule
       when(mockSourceQueue.offer((scheduleId, schedule.toScheduledMessage)))
@@ -46,6 +75,7 @@ class PublisherActorSpec extends AkkaSpecBase with MockitoSugar {
     }
 
     "stop when queue fails" in new TestContext {
+      initQueue
       watch(publisherActor)
 
       publisherActor ! DownstreamFailure(new Exception("boom!"))
@@ -56,11 +86,15 @@ class PublisherActorSpec extends AkkaSpecBase with MockitoSugar {
   }
 
   private class TestContext {
+
     val mockSourceQueue = mock[ScheduleQueue]
-    val publisherActor  = TestActorRef(new PublisherActor)
 
-    when(mockSourceQueue.watchCompletion()).thenReturn(Future.never)
-    publisherActor ! Init(mockSourceQueue)
+    val publisherActor = TestActorRef(new PublisherActor)
+
+    def initQueue = {
+      when(mockSourceQueue.watchCompletion()).thenReturn(Future.never)
+      publisherActor ! Init(mockSourceQueue)
+    }
 
     def generateSchedule: (ScheduleId, ScheduleEvent) =
       (UUID.randomUUID().toString, random[ScheduleEvent])