Repository: softwaremill/elasticmq (scala)

Issue:
Invalid error response for ChangeMessageVisibility batch request
A `ChangeMessageVisibilityBatchRequest` that contains an entry with an invalid receipt handle fails entirely with status code 400 (`InvalidParameterValue; see the SQS docs`), rather than returning partial success containing a `BatchResultErrorEntry` with code `ReceiptHandleIsInvalid` for that particular entry.

`DeleteMessageBatchRequest` behaves as expected.

Pull request:
Fix `ChangeMessageVisibilityBatchRequest` to return a `BatchResultErrorEntry` in case of invalid receipt handles rather than failing the request.

It looks like changes to message visibility used to be done internally using message ids rather than receipt handles. 
Though, similarly to #324, the most recent (valid) receipt handle has to be used to be consistent with the SQS API. Therefore this changes `UpdateVisibilityTimeout` to contain a `DeliveryReceipt` rather than a `MessageId`.

(fixes #632)

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 259409f6e..1e2ea80eb 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
@@ -27,8 +27,8 @@ trait QueueActorMessageOps
     msg match {
       case SendMessage(message) =>
         handleOrRedirectMessage(message, context).send()
-      case UpdateVisibilityTimeout(messageId, visibilityTimeout) =>
-        updateVisibilityTimeout(messageId, visibilityTimeout).send()
+      case UpdateVisibilityTimeout(deliveryReceipt, visibilityTimeout) =>
+        updateVisibilityTimeout(deliveryReceipt, visibilityTimeout).send()
       case ReceiveMessages(visibilityTimeout, count, _, receiveRequestAttemptId) =>
         receiveMessages(visibilityTimeout, count, receiveRequestAttemptId).send()
       case DeleteMessage(deliveryReceipt) =>
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/operations/UpdateVisibilityTimeoutOps.scala b/core/src/main/scala/org/elasticmq/actor/queue/operations/UpdateVisibilityTimeoutOps.scala
index 11c63e516..569a7f8cf 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/operations/UpdateVisibilityTimeoutOps.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/operations/UpdateVisibilityTimeoutOps.scala
@@ -8,42 +8,43 @@ trait UpdateVisibilityTimeoutOps extends Logging {
   this: QueueActorStorage =>
 
   def updateVisibilityTimeout(
-      messageId: MessageId,
+      deliveryReceipt: DeliveryReceipt,
       visibilityTimeout: VisibilityTimeout
-  ): ResultWithEvents[Either[MessageDoesNotExist, Unit]] = {
-    updateNextDelivery(messageId, CommonOperations.computeNextDelivery(visibilityTimeout, queueData, nowProvider))
+  ): ResultWithEvents[Either[InvalidReceiptHandle, Unit]] = {
+    updateNextDelivery(deliveryReceipt, CommonOperations.computeNextDelivery(visibilityTimeout, queueData, nowProvider))
   }
 
   private def updateNextDelivery(
-      messageId: MessageId,
+      deliveryReceipt: DeliveryReceipt,
       newNextDelivery: MillisNextDelivery
-  ): ResultWithEvents[Either[MessageDoesNotExist, Unit]] = {
-    messageQueue.byId.get(messageId.id) match {
-      case Some(internalMessage) =>
+  ): ResultWithEvents[Either[InvalidReceiptHandle, Unit]] = {
+    val msgId = deliveryReceipt.extractId.toString
+
+    messageQueue.byId.get(msgId) match {
+      case Some(msg) if msg.deliveryReceipts.lastOption.contains(deliveryReceipt.receipt) =>
         // Updating
-        val oldNextDelivery = internalMessage.nextDelivery
-        internalMessage.nextDelivery = newNextDelivery.millis
+        val oldNextDelivery = msg.nextDelivery
+        msg.nextDelivery = newNextDelivery.millis
 
         if (newNextDelivery.millis < oldNextDelivery) {
           // We have to re-insert the msg, as another msg with a bigger next delivery may be now before it,
           // so the msg wouldn't be correctly received.
           // (!) This may be slow (!)
-          messageQueue = messageQueue.filterNot(_.id == internalMessage.id)
-          messageQueue += internalMessage
+          messageQueue = messageQueue.filterNot(_.id == msg.id)
+          messageQueue += msg
         }
         // Else:
         // Just increasing the next delivery. Common case. It is enough to increase the value in the object. No need to
         // re-insert the msg into the queue, as it will be reinserted if needed during receiving.
 
-        logger.debug(s"${queueData.name}: Updated next delivery of $messageId to $newNextDelivery")
+        logger.debug(s"${queueData.name}: Updated next delivery of $msgId to $newNextDelivery")
 
         ResultWithEvents.valueWithEvents(
           Right(()),
-          List(QueueEvent.MessageUpdated(queueData.name, internalMessage))
+          List(QueueEvent.MessageUpdated(queueData.name, msg))
         )
-
-      case None =>
-        ResultWithEvents.onlyValue(Left(new MessageDoesNotExist(queueData.name, messageId)))
+      case _ =>
+        ResultWithEvents.onlyValue(Left(new InvalidReceiptHandle(queueData.name, deliveryReceipt.receipt)))
     }
   }
 
diff --git a/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala b/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
index 23c085c15..532c32879 100644
--- a/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
+++ b/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
@@ -41,8 +41,8 @@ case class ClearQueue() extends QueueQueueMsg[Unit]
 
 case class SendMessage(message: NewMessageData) extends QueueMessageMsg[MessageData]
 case class MoveMessage(message: InternalMessage, moveDestination: MoveDestination) extends QueueMessageMsg[Unit]
-case class UpdateVisibilityTimeout(messageId: MessageId, visibilityTimeout: VisibilityTimeout)
-    extends QueueMessageMsg[Either[MessageDoesNotExist, Unit]]
+case class UpdateVisibilityTimeout(deliveryReceipt: DeliveryReceipt, visibilityTimeout: VisibilityTimeout)
+    extends QueueMessageMsg[Either[InvalidReceiptHandle, Unit]]
 case class ReceiveMessages(
     visibilityTimeout: VisibilityTimeout,
     count: Int,
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityBatchDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityBatchDirectives.scala
index 48b4abe18..4f7df9f1a 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityBatchDirectives.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityBatchDirectives.scala
@@ -1,20 +1,34 @@
 package org.elasticmq.rest.sqs
 
 import Constants._
+import org.elasticmq.{DeliveryReceipt, MillisVisibilityTimeout}
+import org.elasticmq.msg.UpdateVisibilityTimeout
+import org.elasticmq.actor.reply._
 import org.elasticmq.rest.sqs.Action.ChangeMessageVisibilityBatch
 import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
 
 trait ChangeMessageVisibilityBatchDirectives {
-  this: ElasticMQDirectives with ChangeMessageVisibilityDirectives with BatchRequestsModule =>
+  this: ElasticMQDirectives with BatchRequestsModule =>
   def changeMessageVisibilityBatch(p: AnyParams) = {
     p.action(ChangeMessageVisibilityBatch) {
       queueActorFromRequest(p) { queueActor =>
         val resultsFuture =
           batchRequest("ChangeMessageVisibilityBatchRequestEntry", p) { (messageData, id, _) =>
-            doChangeMessageVisibility(queueActor, messageData).map { _ =>
-              <ChangeMessageVisibilityBatchResultEntry>
-              <Id>{id}</Id>
-            </ChangeMessageVisibilityBatchResultEntry>
+            val receiptHandle = messageData(ReceiptHandleParameter)
+            val visibilityTimeout = MillisVisibilityTimeout.fromSeconds(messageData(VisibilityTimeoutParameter).toLong)
+            val result = queueActor ? UpdateVisibilityTimeout(DeliveryReceipt(receiptHandle), visibilityTimeout)
+
+            result.map {
+              case Left(error) =>
+                <BatchResultErrorEntry>
+                  <Id>{id}</Id>
+                  <Code>{error.code}</Code>
+                  <Message>{error.message}</Message>
+                </BatchResultErrorEntry>
+              case Right(_) =>
+                <ChangeMessageVisibilityBatchResultEntry>
+                  <Id>{id}</Id>
+                </ChangeMessageVisibilityBatchResultEntry>
             }
           }
 
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityDirectives.scala
index 46d01da35..2fe178631 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityDirectives.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ChangeMessageVisibilityDirectives.scala
@@ -1,41 +1,35 @@
 package org.elasticmq.rest.sqs
 
-import Constants._
-import org.elasticmq.{DeliveryReceipt, MillisVisibilityTimeout}
-import akka.actor.ActorRef
 import org.elasticmq.actor.reply._
 import org.elasticmq.msg.UpdateVisibilityTimeout
 import org.elasticmq.rest.sqs.Action.ChangeMessageVisibility
+import org.elasticmq.rest.sqs.Constants._
 import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
+import org.elasticmq.{DeliveryReceipt, MillisVisibilityTimeout}
 
 trait ChangeMessageVisibilityDirectives { this: ElasticMQDirectives =>
   def changeMessageVisibility(p: AnyParams) = {
     p.action(ChangeMessageVisibility) {
       queueActorFromRequest(p) { queueActor =>
-        doChangeMessageVisibility(queueActor, p).map { _ =>
-          respondWith {
-            <ChangeMessageVisibilityResponse>
-              <ResponseMetadata>
-                <RequestId>{EmptyRequestId}</RequestId>
-              </ResponseMetadata>
-            </ChangeMessageVisibilityResponse>
-          }
+        (p.requiredParam(ReceiptHandleParameter) and p.requiredParam(VisibilityTimeoutParameter)) {
+          (receipt, visibilityTimeout) =>
+            val result = queueActor ? UpdateVisibilityTimeout(
+              DeliveryReceipt(receipt),
+              MillisVisibilityTimeout.fromSeconds(visibilityTimeout.toLong)
+            )
+            result.map {
+              case Left(error) => throw new SQSException(error.code, errorMessage = Some(error.message))
+              case Right(_) =>
+                respondWith {
+                  <ChangeMessageVisibilityResponse>
+                  <ResponseMetadata>
+                    <RequestId>{EmptyRequestId}</RequestId>
+                  </ResponseMetadata>
+                </ChangeMessageVisibilityResponse>
+                }
+            }
         }
       }
     }
   }
-
-  def doChangeMessageVisibility(queueActor: ActorRef, parameters: AnyParams) = {
-    val visibilityTimeout = MillisVisibilityTimeout.fromSeconds(parameters(VisibilityTimeoutParameter).toLong)
-    val msgId = DeliveryReceipt(parameters(ReceiptHandleParameter)).extractId
-
-    for {
-      updateResult <- queueActor ? UpdateVisibilityTimeout(msgId, visibilityTimeout)
-    } yield {
-      updateResult match {
-        case Left(_)  => throw SQSException.invalidParameterValue
-        case Right(_) => // ok
-      }
-    }
-  }
 }


Test patch:
diff --git a/core/src/test/scala/org/elasticmq/actor/QueueActorMsgOpsTest.scala b/core/src/test/scala/org/elasticmq/actor/QueueActorMsgOpsTest.scala
index df5f49504..e0ed81bb6 100644
--- a/core/src/test/scala/org/elasticmq/actor/QueueActorMsgOpsTest.scala
+++ b/core/src/test/scala/org/elasticmq/actor/QueueActorMsgOpsTest.scala
@@ -189,14 +189,15 @@ class QueueActorMsgOpsTest extends ActorTest with QueueManagerForEachTest with D
   test("increasing next delivery of a msg") {
     // Given
     val q1 = createQueueData("q1", MillisVisibilityTimeout(1L))
-    val m = createNewMessageData("xyz", "1234", Map(), MillisNextDelivery(123L))
+    val m = createNewMessageData("xyz", "1234", Map(), MillisNextDelivery(100L))
 
     for {
       Right(queueActor) <- queueManagerActor ? CreateQueue(q1)
       _ <- queueActor ? SendMessage(m)
+      List(mdata) <- queueActor ? ReceiveMessages(DefaultVisibilityTimeout, 1, None, None)
 
       // When
-      _ <- queueActor ? UpdateVisibilityTimeout(m.id.get, MillisVisibilityTimeout(50L))
+      _ <- queueActor ? UpdateVisibilityTimeout(mdata.deliveryReceipt.get, MillisVisibilityTimeout(50L))
       lookupResult <- queueActor ? LookupMessage(MessageId("xyz"))
     } yield {
       // Then
@@ -208,19 +209,21 @@ class QueueActorMsgOpsTest extends ActorTest with QueueManagerForEachTest with D
 
   test("decreasing next delivery of a msg") {
     // Given
-    val q1 = createQueueData("q1", MillisVisibilityTimeout(1L)) // Initially m2 should be delivered after m1
+    val q1 = createQueueData("q1", MillisVisibilityTimeout(100L)) // Initially m2 should be delivered after m1
     val m1 =
-      createNewMessageData("xyz1", "1234", Map(), MillisNextDelivery(150L))
+      createNewMessageData("xyz1", "1234", Map(), MillisNextDelivery(50L))
     val m2 =
-      createNewMessageData("xyz2", "1234", Map(), MillisNextDelivery(200L))
+      createNewMessageData("xyz2", "1234", Map(), MillisNextDelivery(100L))
 
     for {
       Right(queueActor) <- queueManagerActor ? CreateQueue(q1)
       _ <- queueActor ? SendMessage(m1)
       _ <- queueActor ? SendMessage(m2)
 
+      List(_, m2data) <- queueActor ? ReceiveMessages(DefaultVisibilityTimeout, 2, None, None)
+
       // When
-      _ <- queueActor ? UpdateVisibilityTimeout(m2.id.get, MillisVisibilityTimeout(10L))
+      _ <- queueActor ? UpdateVisibilityTimeout(m2data.deliveryReceipt.get, MillisVisibilityTimeout(10L))
       _ = nowProvider.mutableNowMillis.set(110L)
       receiveResult <- queueActor ? ReceiveMessages(DefaultVisibilityTimeout, 1, None, None)
     } yield {
diff --git a/core/src/test/scala/org/elasticmq/actor/QueueEventListenerTest.scala b/core/src/test/scala/org/elasticmq/actor/QueueEventListenerTest.scala
index 26e59f20e..a9c4b5551 100644
--- a/core/src/test/scala/org/elasticmq/actor/QueueEventListenerTest.scala
+++ b/core/src/test/scala/org/elasticmq/actor/QueueEventListenerTest.scala
@@ -53,8 +53,9 @@ class QueueEventListenerTest extends ActorTest with QueueManagerWithListenerForE
   test("QueueMessageUpdated event should be triggerred") {
     for {
       Right(queue) <- queueManagerActor ? CreateQueue(createQueueData("q1", MillisVisibilityTimeout(1L)))
-      msg <- queue ? SendMessage(createNewMessageData("abc", "xyz", Map.empty, MillisNextDelivery(100)))
-      _ <- queue ? UpdateVisibilityTimeout(msg.id, MillisVisibilityTimeout(2000))
+      _ <- queue ? SendMessage(createNewMessageData("abc", "xyz", Map.empty, MillisNextDelivery(100)))
+      List(msg) <- queue ? ReceiveMessages(MillisVisibilityTimeout(1), 1, None, None)
+      _ <- queue ? UpdateVisibilityTimeout(msg.deliveryReceipt.get, MillisVisibilityTimeout(2000))
     } yield {
       queueEventListener.expectMsgType[QueueEvent.QueueCreated].queue.name shouldBe "q1"
       queueEventListener.expectMsgType[QueueEvent.MessageAdded].message.id shouldBe "abc"
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
index bc99a4f80..18a98126c 100644
--- a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
@@ -1168,7 +1168,8 @@ class AmazonJavaSdkTestSuite extends SqsClientServerCommunication with Matchers
         .withQueueUrl(queueUrl)
         .withEntries(
           new ChangeMessageVisibilityBatchRequestEntry("1", msg1.getReceiptHandle).withVisibilityTimeout(1),
-          new ChangeMessageVisibilityBatchRequestEntry("2", msg2.getReceiptHandle).withVisibilityTimeout(1)
+          new ChangeMessageVisibilityBatchRequestEntry("2", msg2.getReceiptHandle).withVisibilityTimeout(1),
+          new ChangeMessageVisibilityBatchRequestEntry("3", "invalid").withVisibilityTimeout(1)
         )
     )
 
@@ -1179,6 +1180,7 @@ class AmazonJavaSdkTestSuite extends SqsClientServerCommunication with Matchers
 
     // Then
     result.getSuccessful.asScala.map(_.getId).toSet should be(Set("1", "2"))
+    result.getFailed.asScala.map(e => (e.getId, e.getCode)) should be(List(("3", "ReceiptHandleIsInvalid")))
 
     Set(m3, m4) should be(Set(Some("Message 1"), Some("Message 2")))
   }
@@ -1712,7 +1714,13 @@ class AmazonJavaSdkTestSuite extends SqsClientServerCommunication with Matchers
     // Given
     val queueUrl = client.createQueue(new CreateQueueRequest("testQueue1")).getQueueUrl
 
-    val handle = client.sendMessage(new SendMessageRequest(queueUrl, "Message 1").withDelaySeconds(4)).getMessageId
+    client.sendMessage(new SendMessageRequest(queueUrl, "Message 1"))
+    val handle = client
+      .receiveMessage(new ReceiveMessageRequest(queueUrl).withVisibilityTimeout(4))
+      .getMessages
+      .get(0)
+      .getReceiptHandle
+
     new Thread {
       override def run(): Unit = {
         Thread.sleep(1000L)