Repository: softwaremill/elasticmq (scala)

Issue:
[BUG] Message Persistence: Deleted dead letter queue messages are recovered after docker container restart
Expected:
---
GIVEN message persistence is enabled 
AND a message moved to a dead letter queue is manually deleted
I EXPECT that a message deleted from the dead letter queue will not be restored after the docker container restarts

Actual:
---
Messages previously deleted from a dead letter queue are restored after `docker restart myelasticmqinstance`


To reproduce:
---
- elasticmq.conf
```
include classpath("application.conf")

# What is the outside visible address of this ElasticMQ node
# Used to create the queue URL (may be different from bind address!)
node-address {
  protocol = http
  host = localhost
  port = 9324
  context-path = ""
}

rest-sqs {
  enabled = true
  bind-port = 9324
  bind-hostname = "0.0.0.0"
  # Possible values: relaxed, strict
  sqs-limits = strict
}

rest-stats {
  enabled = true
  bind-port = 9325
  bind-hostname = "0.0.0.0"
}

# Should the node-address be generated from the bind port/hostname
# Set this to true e.g. when assigning port automatically by using port 0.
generate-node-address = false

queues {
  myqueue {
    defaultVisibilityTimeout = 1 seconds
    delay = 5 seconds
    receiveMessageWait = 0 seconds
    deadLettersQueue {
      name = "myqueue-dead-letters"
      maxReceiveCount = 1 // from 1 to 1000
    }
    contentBasedDeduplication = true
    tags {
      tag1 = "tagged1"
      tag2 = "tagged2"
    }
  }
  myqueue-dead-letters { }
}

queues-storage {
  enabled = true
}

messages-storage {
  enabled = true
}

# Region and accountId which will be included in resource ids
aws {
  region = us-west-2
  accountId = 000000000000
}
```

- Run elasticmq via docker
```
docker run --name myelasticmqinstance  -p 9324:9324 -p 9325:9325 -v `pwd`/conf/elasticmq.conf:/opt/elasticmq.conf -v `pwd`/data:/data softwaremill/elasticmq-native
```

- Send a message to `myqueue`
```
aws --endpoint-url http://localhost:9324 sqs send-message --queue-url http://localhost:9324/queue/myqueue --message-body "Hello, queue!"
```

- View message in `myqueue` twice using command below twice
```
aws --endpoint-url http://localhost:9324 sqs receive-message --queue-url http://localhost:9324/queue/myqueue
```

- Copy the message receipt handle
- Observe the message go to the dead letter queue (`http://localhost:9325`) after 2 views, given config above

- Manually delete the message from `myqueue-dead-letters` queue using the receipt handle saved in previous step
```
aws --endpoint-url http://localhost:9324 sqs delete-message --queue-url http://localhost:9324/queue/myqueue-dead-letters --receipt-handle XYZ...
```

- Stop the docker instance
- Start the docker instance
- View the message queue contents `http://localhost:9325`; observe the deleted message has been restored

Pull request:
fixes #618

Patch:
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
index 1be63e3fc..bd5716d30 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
@@ -34,7 +34,7 @@ trait QueueActorMessageOps
       case DeleteMessage(deliveryReceipt) =>
         deleteMessage(deliveryReceipt).send()
       case LookupMessage(messageId)          => messageQueue.getById(messageId.id).map(_.toMessageData)
-      case MoveMessage(message, destination) => moveMessage(message, destination)
+      case MoveMessage(message, destination) => moveMessage(message, destination).send()
       case DeduplicationIdsCleanup =>
         fifoMessagesHistory = fifoMessagesHistory.cleanOutdatedMessages(nowProvider)
         DoNotReply()
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala
index c22ca431b..32770962f 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala
@@ -33,6 +33,10 @@ trait QueueActorStorage {
   case class ResultWithEvents[T](result: Option[T], events: List[QueueEventWithOperationStatus] = List.empty)
       extends Logging {
 
+    def mapResult[U](f: T => U): ResultWithEvents[U] = {
+      ResultWithEvents(result.map(f), events)
+    }
+
     def send[U](recipient: Option[ActorRef] = None): ReplyAction[U] = {
       val notificationF =
         if (events == Nil || queueEventListener.isEmpty) {
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/operations/MoveMessageOps.scala b/core/src/main/scala/org/elasticmq/actor/queue/operations/MoveMessageOps.scala
index 6fb732529..5bffb6c2e 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/operations/MoveMessageOps.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/operations/MoveMessageOps.scala
@@ -1,6 +1,6 @@
 package org.elasticmq.actor.queue.operations
 
-import org.elasticmq.actor.queue.{InternalMessage, QueueActorStorage}
+import org.elasticmq.actor.queue.{InternalMessage, QueueActorStorage, QueueEvent}
 import org.elasticmq.msg.SendMessage
 import org.elasticmq.util.Logging
 import org.elasticmq.{DeduplicationId, MoveDestination, MoveToDLQ}
@@ -8,7 +8,7 @@ import org.elasticmq.{DeduplicationId, MoveDestination, MoveToDLQ}
 trait MoveMessageOps extends Logging {
   this: QueueActorStorage =>
 
-  def moveMessage(message: InternalMessage, destination: MoveDestination): Unit = {
+  def moveMessage(message: InternalMessage, destination: MoveDestination): ResultWithEvents[Unit] = {
 
     copyMessagesToActorRef.foreach { _ ! SendMessage(message.toNewMessageData) }
 
@@ -16,7 +16,7 @@ trait MoveMessageOps extends Logging {
       case MoveToDLQ =>
         if (queueData.isFifo) {
           CommonOperations.wasRegistered(message.toNewMessageData, fifoMessagesHistory) match {
-            case Some(_) => ()
+            case Some(_) => ResultWithEvents.empty
             case None =>
               logger.debug(s"Moved message (${message.id}) from FIFO queue to ${queueData.name}")
               moveMessageToQueue(regenerateDeduplicationId(message))
@@ -34,8 +34,10 @@ trait MoveMessageOps extends Logging {
     )
   }
 
-  private def moveMessageToQueue(internalMessage: InternalMessage): Unit = {
+  private def moveMessageToQueue(internalMessage: InternalMessage): ResultWithEvents[Unit] = {
     messageQueue += internalMessage
     fifoMessagesHistory = fifoMessagesHistory.addNew(internalMessage)
+
+    ResultWithEvents.onlyEvents(List(QueueEvent.MessageAdded(queueData.name, internalMessage)))
   }
 }
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/operations/ReceiveMessageOps.scala b/core/src/main/scala/org/elasticmq/actor/queue/operations/ReceiveMessageOps.scala
index 19512f806..6b9402016 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/operations/ReceiveMessageOps.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/operations/ReceiveMessageOps.scala
@@ -9,43 +9,66 @@ import org.elasticmq.util.{Logging, NowProvider}
 trait ReceiveMessageOps extends Logging {
   this: QueueActorStorage with DeleteMessageOps =>
 
+  trait MessageWithDestination
+  case class MessageToReturn(internalMessage: InternalMessage) extends MessageWithDestination
+  case class MessageToDelete(internalMessage: InternalMessage) extends MessageWithDestination
+
   protected def receiveMessages(
       visibilityTimeout: VisibilityTimeout,
       count: Int,
       receiveRequestAttemptId: Option[String]
   ): ResultWithEvents[List[MessageData]] = {
     implicit val np: NowProvider = nowProvider
-    val messages = receiveRequestAttemptId
-      .flatMap({ attemptId =>
-        // for a given request id, check for any messages we've dequeued and cached
-        val cachedMessages = getMessagesFromRequestAttemptCache(attemptId)
 
-        // if the cache returns an empty list instead of None, we still want to pull messages from
-        // from the queue so return None in that case to properly process down stream
-        cachedMessages.getOrElse(Nil) match {
-          case Nil     => None
-          case default => Some(default)
-        }
-      })
+    val messagesWithDestinations = receiveRequestAttemptId
+      .flatMap({ attemptId => getCachedMessages(attemptId) })
       .getOrElse(getMessagesFromQueue(count))
-      .map { internalMessage =>
-        // Putting the msg again into the queue, with a new next delivery
-        val newNextDelivery = CommonOperations.computeNextDelivery(visibilityTimeout, queueData, np)
-        internalMessage.trackDelivery(newNextDelivery)
-        messageQueue += internalMessage
 
-        logger.debug(s"${queueData.name}: Receiving message ${internalMessage.id}")
-        internalMessage
-      }
+    val messagesToReturn = messagesWithDestinations.flatMap {
+      case MessageToReturn(internalMessage) => Some(internalMessage)
+      case _                                => None
+    } map { internalMessage =>
+      // Putting the msg again into the queue, with a new next delivery
+      val newNextDelivery = CommonOperations.computeNextDelivery(visibilityTimeout, queueData, np)
+      internalMessage.trackDelivery(newNextDelivery)
+      messageQueue += internalMessage
+
+      logger.debug(s"${queueData.name}: Receiving message ${internalMessage.id}")
+      internalMessage
+    }
+
+    val updateEvents =
+      messagesToReturn.map(internalMessage => QueueEvent.MessageUpdated(queueData.name, internalMessage))
 
-    receiveRequestAttemptId.foreach { attemptId => receiveRequestAttemptCache.add(attemptId, messages) }
+    val deleteEvents = messagesWithDestinations.flatMap {
+      case MessageToDelete(internalMessage) => Some(internalMessage)
+      case _                                => None
+    } flatMap { internalMessage =>
+      internalMessage.deliveryReceipts.toList.map(dr => deleteMessage(DeliveryReceipt(dr))).flatMap(_.events)
+    }
+
+    receiveRequestAttemptId.foreach { attemptId => receiveRequestAttemptCache.add(attemptId, messagesToReturn) }
 
     ResultWithEvents.valueWithEvents(
-      messages.map(_.toMessageData),
-      messages.map(internalMessage => QueueEvent.MessageUpdated(queueData.name, internalMessage))
+      messagesToReturn.map(_.toMessageData),
+      updateEvents ++ deleteEvents
     )
   }
 
+  private def getCachedMessages(
+      attemptId: String
+  )(implicit np: NowProvider): Option[List[MessageWithDestination]] = {
+    // for a given request id, check for any messages we've dequeued and cached
+    val cachedMessages = getMessagesFromRequestAttemptCache(attemptId)
+
+    // if the cache returns an empty list instead of None, we still want to pull messages from
+    // from the queue so return None in that case to properly process down stream
+    cachedMessages.getOrElse(Nil) match {
+      case Nil     => None
+      case default => Some(default.map(internalMessage => MessageToReturn(internalMessage)))
+    }
+  }
+
   private def getMessagesFromRequestAttemptCache(
       receiveRequestAttemptId: String
   )(implicit np: NowProvider): Option[List[InternalMessage]] = {
@@ -57,16 +80,15 @@ trait ReceiveMessageOps extends Logging {
     }
   }
 
-  private def getMessagesFromQueue(count: Int) = {
+  private def getMessagesFromQueue(count: Int): List[MessageWithDestination] = {
     val deliveryTime = nowProvider.nowMillis
-    messageQueue.dequeue(count, deliveryTime).flatMap { internalMessage =>
+    messageQueue.dequeue(count, deliveryTime).map { internalMessage =>
       if (queueData.deadLettersQueue.map(_.maxReceiveCount).exists(_ <= internalMessage.receiveCount)) {
         logger.debug(s"${queueData.name}: send message $internalMessage to dead letters actor $deadLettersActorRef")
         deadLettersActorRef.foreach(_ ! MoveMessage(internalMessage, MoveToDLQ))
-        internalMessage.deliveryReceipts.foreach(dr => deleteMessage(DeliveryReceipt(dr)))
-        None
+        MessageToDelete(internalMessage)
       } else {
-        Some(internalMessage)
+        MessageToReturn(internalMessage)
       }
     }
   }
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala b/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala
index db342d979..43c951714 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala
@@ -32,7 +32,7 @@ trait SendMessageOp extends Logging {
 
   def restoreMessages(messages: List[InternalMessage]): Unit = {
     messages.foreach(addInternalMessage)
-    logger.info(s"Restored ${messages.size} messages")
+    logger.info(s"Restored ${messages.size} messages in queue ${queueData.name}")
   }
 
   private def sendMessage(message: NewMessageData): ResultWithEvents[MessageData] = {


Test patch:
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/SqlQueuePersistenceTest.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/SqlQueuePersistenceTest.scala
index c3639977f..3e1860ddc 100644
--- a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/SqlQueuePersistenceTest.scala
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/SqlQueuePersistenceTest.scala
@@ -1,13 +1,20 @@
 package org.elasticmq.rest.sqs
 
-import com.amazonaws.services.sqs.model.{CreateQueueRequest, ReceiveMessageRequest, SendMessageRequest}
+import com.amazonaws.services.sqs.model.{
+  CreateQueueRequest,
+  GetQueueAttributesRequest,
+  ReceiveMessageRequest,
+  SendMessageRequest
+}
 import org.elasticmq.actor.reply._
 import org.elasticmq.persistence.sql.GetAllMessages
+import org.elasticmq.rest.sqs.model.RedrivePolicy
 import org.elasticmq.util.Logging
 import org.scalatest.BeforeAndAfter
-import org.scalatest.concurrent.ScalaFutures.convertScalaFuture
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
+import org.elasticmq.rest.sqs.model.RedrivePolicyJson.format
+import spray.json.enrichAny
 
 import scala.collection.JavaConverters._
 
@@ -19,6 +26,10 @@ class SqlQueuePersistenceTest
     with Logging {
 
   val testQueueName = "testQueue1"
+  val deadLetterQueueName = "testDLQ1"
+
+  val awsAccountId = "123456789012"
+  val awsRegion = "elasticmq"
 
   test("should persist the messages and after restart read the messages") {
     startServerAndRun(pruneDataOnInit = true) {
@@ -142,4 +153,90 @@ class SqlQueuePersistenceTest
       bodies2 should be(Set("Message 1", "Message 2", "Message 3"))
     }
   }
+
+  test("should move messages to DLQ and restore") {
+    startServerAndRun(pruneDataOnInit = true) {
+      val dlQueueUrl = client
+        .createQueue(
+          new CreateQueueRequest(deadLetterQueueName)
+        )
+        .getQueueUrl
+
+      val queueUrl = client
+        .createQueue(
+          new CreateQueueRequest(testQueueName)
+            .withAttributes(
+              Map(
+                "VisibilityTimeout" -> "1",
+                "RedrivePolicy" -> RedrivePolicy(deadLetterQueueName, awsAccountId, awsRegion, 1).toJson.toString
+              ).asJava
+            )
+        )
+        .getQueueUrl
+
+      client.sendMessage(new SendMessageRequest(queueUrl, "Message 1"))
+      client.sendMessage(new SendMessageRequest(queueUrl, "Message 2"))
+      client.sendMessage(new SendMessageRequest(queueUrl, "Message 3"))
+
+      val messages1 = client.receiveMessage(new ReceiveMessageRequest(queueUrl).withMaxNumberOfMessages(3)).getMessages
+      messages1 should have size 3
+
+      Thread.sleep(1000)
+
+      val messages2 = client.receiveMessage(new ReceiveMessageRequest(queueUrl).withMaxNumberOfMessages(3)).getMessages
+      messages2 shouldBe empty
+
+      val attributes =
+        client
+          .getQueueAttributes(new GetQueueAttributesRequest(queueUrl).withAttributeNames("ApproximateNumberOfMessages"))
+          .getAttributes
+          .asScala
+
+      attributes.get("ApproximateNumberOfMessages") shouldBe Some("0")
+
+      val dlqAttributes =
+        client
+          .getQueueAttributes(
+            new GetQueueAttributesRequest(dlQueueUrl).withAttributeNames("ApproximateNumberOfMessages")
+          )
+          .getAttributes
+          .asScala
+
+      dlqAttributes.get("ApproximateNumberOfMessages") shouldBe Some("3")
+    }
+
+    Thread.sleep(1000)
+
+    startServerAndRun(pruneDataOnInit = false) {
+      val queueUrl = client.getQueueUrl(testQueueName).getQueueUrl
+      val dlQueueUrl = client.getQueueUrl(deadLetterQueueName).getQueueUrl
+
+      val attributes =
+        client
+          .getQueueAttributes(new GetQueueAttributesRequest(queueUrl).withAttributeNames("ApproximateNumberOfMessages"))
+          .getAttributes
+          .asScala
+
+      attributes.get("ApproximateNumberOfMessages") shouldBe Some("0")
+
+      val dlqAttributes =
+        client
+          .getQueueAttributes(
+            new GetQueueAttributesRequest(dlQueueUrl).withAttributeNames("ApproximateNumberOfMessages")
+          )
+          .getAttributes
+          .asScala
+
+      dlqAttributes.get("ApproximateNumberOfMessages") shouldBe Some("3")
+
+      val messages = client.receiveMessage(new ReceiveMessageRequest(queueUrl).withMaxNumberOfMessages(3)).getMessages
+      messages shouldBe empty
+
+      val dlqMessages =
+        client.receiveMessage(new ReceiveMessageRequest(dlQueueUrl).withMaxNumberOfMessages(3)).getMessages
+
+      val bodies = dlqMessages.asScala.map(_.getBody).toSet
+      bodies shouldBe Set("Message 1", "Message 2", "Message 3")
+    }
+  }
 }