Merge "[AMQP 1.0] Resend messages that are released or modified"

4 files changed, 192 insertions, 70 deletions

diff --git a/oslo_messaging/_drivers/amqp1_driver/controller.py b/oslo_messaging/_drivers/amqp1_driver/controller.py
index 49aba92..56fd977 100644
--- a/oslo_messaging/_drivers/amqp1_driver/controller.py
+++ b/oslo_messaging/_drivers/amqp1_driver/controller.py
@@ -104,10 +104,10 @@ class SendTask(Task):
         self.target = target() if isinstance(target, Target) else target
         self.message = message
         self.deadline = deadline
-        self.retry = retry
         self.wait_for_ack = wait_for_ack
         self.service = SERVICE_NOTIFY if notification else SERVICE_RPC
         self.timer = None
+        self._retry = None if retry is None or retry < 0 else retry
         self._wakeup = threading.Event()
         self._error = None
 
@@ -122,18 +122,15 @@ class SendTask(Task):
         """Called immediately before the message is handed off to the i/o
         system.  This implies that the sender link is up.
         """
-        if not self.wait_for_ack:
-            # sender is not concerned with waiting for acknowledgment
-            # "best effort at-most-once delivery"
-            self._cleanup()
-            self._wakeup.set()
+        pass
 
     def _on_ack(self, state, info):
-        """Called by eventloop thread when the ack/nack is received from the
-        peer.
+        """If wait_for_ack is True, this is called by the eventloop thread when
+        the ack/nack is received from the peer.  If wait_for_ack is False this
+        is called by the eventloop right after the message is written to the
+        link.  In the last case state will always be set to ACCEPTED.
         """
         if state != pyngus.SenderLink.ACCEPTED:
-            # TODO(kgiusti): could retry if deadline not hit
             msg = ("{name} message send to {target} failed: remote"
                    " disposition: {disp}, info:"
                    "{info}".format(name=self.name,
@@ -179,15 +176,23 @@ class SendTask(Task):
             self.timer.cancel()
             self.timer = None
 
+    @property
+    def _can_retry(self):
+        # has the retry count expired?
+        if self._retry is not None:
+            self._retry -= 1
+            if self._retry < 0:
+                return False
+        return True
+
 
 class RPCCallTask(SendTask):
     """Performs an RPC Call.  Sends the request and waits for a response from
     the destination.
     """
-
-    def __init__(self, target, message, deadline, retry, wait_for_ack):
+    def __init__(self, target, message, deadline, retry):
         super(RPCCallTask, self).__init__("RPC Call", message, target,
-                                          deadline, retry, wait_for_ack)
+                                          deadline, retry, wait_for_ack=True)
         self._reply_link = None
         self._reply_msg = None
         self._msg_id = None
@@ -198,32 +203,30 @@ class RPCCallTask(SendTask):
 
     def _prepare(self, sender):
         # reserve a message id for mapping the received response
+        if self._msg_id:
+            # already set so this is a re-transmit. To be safe cancel the old
+            # msg_id and allocate a fresh one.
+            self._reply_link.cancel_response(self._msg_id)
         self._reply_link = sender._reply_link
         rl = self._reply_link
         self._msg_id = rl.prepare_for_response(self.message, self._on_reply)
 
     def _on_reply(self, message):
         # called if/when the reply message arrives
-        if self._wakeup.is_set():
-            LOG.debug("RPC Reply received after call completed")
-            return
         self._reply_msg = message
-        self._reply_link = None
+        self._msg_id = None  # to prevent _cleanup() from cancelling it
         self._cleanup()
         self._wakeup.set()
 
     def _on_ack(self, state, info):
-        if self._wakeup.is_set():
-            LOG.debug("RPC ACKed after call completed: %s %s", state, info)
-            return
         if state != pyngus.SenderLink.ACCEPTED:
             super(RPCCallTask, self)._on_ack(state, info)
         # must wait for reply if ACCEPTED
 
     def _cleanup(self):
-        if self._reply_link and self._msg_id:
+        if self._msg_id:
             self._reply_link.cancel_response(self._msg_id)
-            self._msg_id = None
+        self._reply_link = None
         super(RPCCallTask, self)._cleanup()
 
 
@@ -260,18 +263,23 @@ class Sender(pyngus.SenderEventHandler):
         self._address = None
         self._link = None
         self._scheduler = scheduler
-        self._delay = delay  # for re-connecting
+        self._delay = delay  # for re-connecting/re-transmitting
         # holds all pending SendTasks
         self._pending_sends = collections.deque()
         # holds all messages sent but not yet acked
         self._unacked = set()
         self._reply_link = None
         self._connection = None
+        self._resend_timer = None
 
     @property
     def pending_messages(self):
         return len(self._pending_sends)
 
+    @property
+    def unacked_messages(self):
+        return len(self._unacked)
+
     def attach(self, connection, reply_link, addresser):
         """Open the link. Called by the Controller when the AMQP connection
         becomes active.
@@ -290,6 +298,9 @@ class Sender(pyngus.SenderEventHandler):
         LOG.debug("Sender %s detached", self._address)
         self._connection = None
         self._reply_link = None
+        if self._resend_timer:
+            self._resend_timer.cancel()
+            self._resend_timer = None
         if self._link:
             self._link.close()
 
@@ -376,11 +387,9 @@ class Sender(pyngus.SenderEventHandler):
         # sends that have exhausted their retry count:
         expired = set()
         for send_task in self._pending_sends:
-            if send_task.retry is not None:
-                send_task.retry -= 1
-                if send_task.retry <= 0:
-                    expired.add(send_task)
-                    send_task._on_error("Message send failed: %s" % reason)
+            if not send_task._can_retry:
+                expired.add(send_task)
+                send_task._on_error("Message send failed: %s" % reason)
         while expired:
             self._pending_sends.remove(expired.pop())
 
@@ -401,26 +410,75 @@ class Sender(pyngus.SenderEventHandler):
     def _can_send(self):
         return self._link and self._link.active
 
+    # acknowledge status
+    _TIMED_OUT = pyngus.SenderLink.TIMED_OUT
+    _ACCEPTED = pyngus.SenderLink.ACCEPTED
+    _RELEASED = pyngus.SenderLink.RELEASED
+    _MODIFIED = pyngus.SenderLink.MODIFIED
+
     def _send(self, send_task):
         send_task._prepare(self)
         send_task.message.address = self._address
+        if send_task.wait_for_ack:
+            self._unacked.add(send_task)
+
+            def pyngus_callback(link, handle, state, info):
+                # invoked when the message bus (n)acks this message
+                if state == Sender._TIMED_OUT:
+                    # ignore pyngus timeout - we maintain our own timer
+                    # which will properly deal with this case
+                    return
+                self._unacked.discard(send_task)
+                if state == Sender._ACCEPTED:
+                    send_task._on_ack(Sender._ACCEPTED, info)
+                elif (state == Sender._RELEASED
+                      or (state == Sender._MODIFIED and
+                          # assuming delivery-failed means in-doubt:
+                          not info.get("delivery-failed") and
+                          not info.get("undeliverable-here"))):
+                    # These states indicate that the message was never
+                    # forwarded beyond the next hop so they can be
+                    # re-transmitted without risk of duplication
+                    self._resend(send_task)
+                else:
+                    # some error - let task figure it out...
+                    send_task._on_ack(state, info)
+
+            self._link.send(send_task.message,
+                            delivery_callback=pyngus_callback,
+                            handle=self,
+                            deadline=send_task.deadline)
+        else:  # do not wait for ack
+            self._link.send(send_task.message,
+                            delivery_callback=None,
+                            handle=self,
+                            deadline=send_task.deadline)
+            send_task._on_ack(pyngus.SenderLink.ACCEPTED, {})
+
+    def _resend(self, send_task):
+        # the message bus returned the message without forwarding it. Wait a
+        # bit for other outstanding sends to finish - most likely ending up
+        # here since they are all going to the same destination - then resend
+        # this message
+        if send_task._can_retry:
+            # note well: once there is something on the pending list no further
+            # messages will be sent (they will all queue up behind this one).
+            self._pending_sends.append(send_task)
+            if self._resend_timer is None:
+                sched = self._scheduler
+                # this will get the pending sends going again
+                self._resend_timer = sched.defer(self._resend_pending,
+                                                 self._delay)
+        else:
+            send_task._on_error("Send retries exhausted")
 
-        def pyngus_callback(link, handle, state, info):
-            # invoked when the message bus (n)acks this message
-            if state == pyngus.SenderLink.TIMED_OUT:
-                # ignore pyngus timeout - we maintain our own timer
-                return
-            self._unacked.discard(send_task)
-            send_task._on_ack(state, info)
-
-        self._unacked.add(send_task)
-        self._link.send(send_task.message,
-                        delivery_callback=pyngus_callback,
-                        handle=self,
-                        deadline=send_task.deadline)
+    def _resend_pending(self):
+        # run from the _resend_timer, attempt to resend pending messages
+        self._resend_timer = None
+        self._send_pending()
 
     def _send_pending(self):
-        # send all pending messages
+        # flush all pending messages out
         if self._can_send:
             while self._pending_sends:
                 self._send(self._pending_sends.popleft())
@@ -472,7 +530,7 @@ class Replies(pyngus.ReceiverEventHandler):
             self._receiver.close()
 
     def destroy(self):
-        self._correlation = None
+        self._correlation.clear()
         if self._receiver:
             self._receiver.destroy()
             self._receiver = None
@@ -494,11 +552,10 @@ class Replies(pyngus.ReceiverEventHandler):
         """Abort waiting for the response message corresponding to msg_id.
         This can be used if the request fails and no reply is expected.
         """
-        if self._correlation:
-            try:
-                del self._correlation[msg_id]
-            except KeyError:
-                pass
+        try:
+            del self._correlation[msg_id]
+        except KeyError:
+            pass
 
     @property
     def active(self):
@@ -864,8 +921,6 @@ class Controller(pyngus.ConnectionEventHandler):
         if send_task.deadline and send_task.deadline <= now():
             send_task._on_timeout()
             return
-        if send_task.retry is None or send_task.retry < 0:
-            send_task.retry = None
         key = keyify(send_task.target, send_task.service)
         sender = self._all_senders.get(key)
         if not sender:
@@ -1142,7 +1197,7 @@ class Controller(pyngus.ConnectionEventHandler):
         self._active_senders.clear()
         unused = []
         for key, sender in iteritems(self._all_senders):
-            # clean up any unused sender links
+            # clean up any sender links that no longer have messages to send
             if sender.pending_messages == 0:
                 unused.append(key)
             else:
@@ -1183,7 +1238,7 @@ class Controller(pyngus.ConnectionEventHandler):
             purge = set(self._all_senders.keys()) - self._active_senders
             for key in purge:
                 sender = self._all_senders[key]
-                if sender.pending_messages == 0:
+                if not sender.pending_messages and not sender.unacked_messages:
                     sender.detach()
                     self._purged_senders.append(self._all_senders.pop(key))
             self._active_senders.clear()
diff --git a/oslo_messaging/_drivers/amqp1_driver/opts.py b/oslo_messaging/_drivers/amqp1_driver/opts.py
index 9278ebd..127dd0d 100644
--- a/oslo_messaging/_drivers/amqp1_driver/opts.py
+++ b/oslo_messaging/_drivers/amqp1_driver/opts.py
@@ -109,11 +109,16 @@ amqp1_opts = [
                help='Time to pause between re-connecting an AMQP 1.0 link that'
                ' failed due to a recoverable error.'),
 
+    cfg.IntOpt('default_reply_retry',
+               default=0,
+               min=-1,
+               help='The maximum number of attempts to re-send a reply message'
+               ' which failed due to a recoverable error.'),
+
     cfg.IntOpt('default_reply_timeout',
                default=30,
                min=5,
-               help='The deadline for an rpc reply message delivery.'
-               ' Only used when caller does not provide a timeout expiry.'),
+               help='The deadline for an rpc reply message delivery.'),
 
     cfg.IntOpt('default_send_timeout',
                default=30,
diff --git a/oslo_messaging/_drivers/impl_amqp1.py b/oslo_messaging/_drivers/impl_amqp1.py
index e48c478..44b99ff 100644
--- a/oslo_messaging/_drivers/impl_amqp1.py
+++ b/oslo_messaging/_drivers/impl_amqp1.py
@@ -109,7 +109,7 @@ class ProtonIncomingMessage(base.RpcIncomingMessage):
             task = controller.SendTask("RPC Reply", response, self._reply_to,
                                        # analogous to kombu missing dest t/o:
                                        deadline,
-                                       retry=0,
+                                       retry=driver._default_reply_retry,
                                        wait_for_ack=ack)
             driver._ctrl.add_task(task)
             rc = task.wait()
@@ -216,6 +216,7 @@ class ProtonDriver(base.BaseDriver):
         self._default_reply_timeout = opt_name.default_reply_timeout
         self._default_send_timeout = opt_name.default_send_timeout
         self._default_notify_timeout = opt_name.default_notify_timeout
+        self._default_reply_retry = opt_name.default_reply_retry
 
         # which message types should be sent pre-settled?
         ps = [s.lower() for s in opt_name.pre_settled]
@@ -301,8 +302,7 @@ class ProtonDriver(base.BaseDriver):
             expire = compute_timeout(self._default_send_timeout)
         if wait_for_reply:
             ack = not self._pre_settle_call
-            task = controller.RPCCallTask(target, request, expire, retry,
-                                          wait_for_ack=ack)
+            task = controller.RPCCallTask(target, request, expire, retry)
         else:
             ack = not self._pre_settle_cast
             task = controller.SendTask("RPC Cast", request, target, expire,
diff --git a/oslo_messaging/tests/drivers/test_amqp_driver.py b/oslo_messaging/tests/drivers/test_amqp_driver.py
index 12db37a..30b796b 100644
--- a/oslo_messaging/tests/drivers/test_amqp_driver.py
+++ b/oslo_messaging/tests/drivers/test_amqp_driver.py
@@ -288,7 +288,7 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
         driver.cleanup()
 
     def test_send_timeout(self):
-        """Verify send timeout."""
+        """Verify send timeout - no reply sent."""
         driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
         target = oslo_messaging.Target(topic="test-topic")
         listener = _ListenerThread(
@@ -310,17 +310,19 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
         driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
         target = oslo_messaging.Target(topic="no listener")
 
-        # the broker will send a nack:
+        # the broker will send a nack (released) since there is no active
+        # listener for the target:
         self.assertRaises(oslo_messaging.MessageDeliveryFailure,
                           driver.send, target,
                           {"context": "whatever"},
                           {"method": "drop"},
                           wait_for_reply=True,
+                          retry=0,
                           timeout=1.0)
         driver.cleanup()
 
     def test_send_not_acked(self):
-        """Verify exception thrown if send Nacked."""
+        """Verify exception thrown ack dropped."""
         self.config(pre_settled=[],
                     group="oslo_messaging_amqp")
         driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
@@ -333,7 +335,8 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
                           driver.send, target,
                           {"context": "whatever"},
                           {"method": "drop"},
-                          wait_for_reply=False)
+                          retry=0,
+                          wait_for_reply=True)
         driver.cleanup()
 
     def test_no_ack_cast(self):
@@ -393,7 +396,7 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
         driver.cleanup()
 
     def test_call_failed_reply(self):
-        """Send back an exception"""
+        """Send back an exception generated at the listener"""
         class _FailedResponder(_ListenerThread):
             def __init__(self, listener):
                 super(_FailedResponder, self).__init__(listener, 1)
@@ -434,7 +437,7 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
                 self.started.set()
                 while not self._done:
                     for in_msg in self.listener.poll(timeout=0.5):
-                        # reply will never be acked:
+                        # reply will never be acked (simulate drop):
                         in_msg._reply_to = "!no-ack!"
                         in_msg.reply(reply={'correlation-id':
                                             in_msg.message.get("id")})
@@ -458,6 +461,7 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
 
     def test_listener_requeue(self):
         "Emulate Server requeue on listener incoming messages"
+        self.config(pre_settled=[], group="oslo_messaging_amqp")
         driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
         driver.require_features(requeue=True)
         target = oslo_messaging.Target(topic="test-topic")
@@ -472,10 +476,6 @@ class TestAmqpSend(_AmqpBrokerTestCaseAuto):
         listener.join(timeout=30)
         self.assertFalse(listener.isAlive())
 
-        for x in listener.get_messages():
-            x.requeue()
-            self.assertEqual(x.message, {"msg": "value"})
-
         predicate = lambda: (self._broker.sender_link_requeue_count == 1)
         _wait_until(predicate, 30)
         self.assertTrue(predicate())
@@ -575,7 +575,7 @@ class TestAmqpNotification(_AmqpBrokerTestCaseAuto):
                 try:
                     driver.send_notification(oslo_messaging.Target(topic=t),
                                              "context", {'target': t},
-                                             version)
+                                             version, retry=0)
                 except oslo_messaging.MessageDeliveryFailure:
                     excepted_targets.append(t)
 
@@ -592,15 +592,18 @@ class TestAmqpNotification(_AmqpBrokerTestCaseAuto):
         driver.cleanup()
 
     def test_released_notification(self):
+        """Broker sends a Nack (released)"""
         driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
         self.assertRaises(oslo_messaging.MessageDeliveryFailure,
                           driver.send_notification,
                           oslo_messaging.Target(topic="bad address"),
                           "context", {'target': "bad address"},
-                          2.0)
+                          2.0,
+                          retry=0)
         driver.cleanup()
 
     def test_notification_not_acked(self):
+        """Simulate drop of ack from broker"""
         driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
         # set this directly so we can use a value < minimum allowed
         driver._default_notify_timeout = 2
@@ -608,7 +611,7 @@ class TestAmqpNotification(_AmqpBrokerTestCaseAuto):
                           driver.send_notification,
                           oslo_messaging.Target(topic="!no-ack!"),
                           "context", {'target': "!no-ack!"},
-                          2.0)
+                          2.0, retry=0)
         driver.cleanup()
 
     def test_no_ack_notification(self):
@@ -1388,6 +1391,64 @@ class TestAddressing(test_utils.BaseTestCase):
                               LegacyAddresser)
 
 
+@testtools.skipUnless(pyngus, "proton modules not present")
+class TestMessageRetransmit(_AmqpBrokerTestCase):
+    # test message is retransmitted if safe to do so
+    def _test_retransmit(self, nack_method):
+        self._nack_count = 2
+
+        def _on_message(message, handle, link):
+            if self._nack_count:
+                self._nack_count -= 1
+                nack_method(link, handle)
+            else:
+                self._broker.forward_message(message, handle, link)
+
+        self._broker.on_message = _on_message
+        self._broker.start()
+        self.config(link_retry_delay=1, pre_settled=[],
+                    group="oslo_messaging_amqp")
+        driver = amqp_driver.ProtonDriver(self.conf, self._broker_url)
+        target = oslo_messaging.Target(topic="test-topic")
+        listener = _ListenerThread(driver.listen(target,
+                                                 None,
+                                                 None)._poll_style_listener,
+                                   1)
+        rc = driver.send(target, {"context": "whatever"},
+                         {"method": "echo", "id": "blah"},
+                         wait_for_reply=True,
+                         retry=2)  # initial send + up to 2 resends
+        self.assertIsNotNone(rc)
+        self.assertEqual(0, self._nack_count)
+        self.assertEqual(rc.get('correlation-id'), 'blah')
+        listener.join(timeout=30)
+        self.assertFalse(listener.isAlive())
+        driver.cleanup()
+
+    def test_released(self):
+        # should retry and succeed
+        self._test_retransmit(lambda l, h: l.message_released(h))
+
+    def test_modified(self):
+        # should retry and succeed
+        self._test_retransmit(lambda l, h: l.message_modified(h,
+                                                              False,
+                                                              False,
+                                                              {}))
+
+    def test_modified_failed(self):
+        # since delivery_failed is set to True, should fail
+        self.assertRaises(oslo_messaging.MessageDeliveryFailure,
+                          self._test_retransmit,
+                          lambda l, h: l.message_modified(h, True, False, {}))
+
+    def test_rejected(self):
+        # rejected - should fail
+        self.assertRaises(oslo_messaging.MessageDeliveryFailure,
+                          self._test_retransmit,
+                          lambda l, h: l.message_rejected(h, {}))
+
+
 class FakeBroker(threading.Thread):
     """A test AMQP message 'broker'."""
 
@@ -1609,7 +1670,7 @@ class FakeBroker(threading.Thread):
 
             def message_received(self, receiver_link, message, handle):
                 """Forward this message out the proper sending link."""
-                self.server.forward_message(message, handle, receiver_link)
+                self.server.on_message(message, handle, receiver_link)
                 if self.link.capacity < 1:
                     self.server.on_credit_exhausted(self.link)
 
@@ -1674,6 +1735,7 @@ class FakeBroker(threading.Thread):
         self.on_sender_active = lambda link: None
         self.on_receiver_active = lambda link: link.add_capacity(10)
         self.on_credit_exhausted = lambda link: link.add_capacity(10)
+        self.on_message = lambda m, h, l: self.forward_message(m, h, l)
 
     def start(self):
         """Start the server."""