# message\_ack failing and retriggering task

**URL:** <https://discourse.nameko.io/t/message-ack-failing-and-retriggering-task/191>\
**Category:** googlegroup\
**Created:** [April 22, 2017, 9:42pm UTC](https://discourse.nameko.io/t/message-ack-failing-and-retriggering-task/191 "2017-04-22T21:42:23Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Diego](https://avatars.discourse-cdn.com/v4/letter/d/f04885/32.png) [@Diego](https://discourse.nameko.io/u/Diego)\
**Post date:** [April 22, 2017, 9:42pm UTC](https://discourse.nameko.io/t/message-ack-failing-and-retriggering-task/191/1 "2017-04-22T21:42:23Z")

</div>

Hi,

I have an rpc endpoint that does some processing and takes around 20 min to  
complete. It basically kicks off a bunch of other async calls to other rpc  
endpoints and waits for all of those to complete.  
I frequently make async calls to this endpoint to do some processing.

However, I get this error message:  
Traceback (most recent call last):  
&nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 177, in run  
&nbsp;&nbsp;&nbsp;&nbsp;for \_ in self.consume(limit=None): # pragma: no cover  
&nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 197, in consume  
&nbsp;&nbsp;&nbsp;&nbsp;self.on\_iteration()  
&nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 417, in on\_iteration  
&nbsp;&nbsp;&nbsp;&nbsp;self.\_process\_pending\_message\_acks()  
&nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 368, in  
\_process\_pending\_message\_acks  
&nbsp;&nbsp;&nbsp;&nbsp;msg.ack()  
&nbsp;&nbsp;File "site-packages/kombu/message.py", line 88, in ack  
&nbsp;&nbsp;&nbsp;&nbsp;self.channel.basic\_ack(self.delivery\_tag)  
&nbsp;&nbsp;File "site-packages/amqp/channel.py", line 1584, in basic\_ack  
&nbsp;&nbsp;&nbsp;&nbsp;self.\_send\_method((60, 80), args)  
&nbsp;&nbsp;File "site-packages/amqp/abstract\_channel.py", line 50, in \_send\_method  
&nbsp;&nbsp;&nbsp;&nbsp;raise RecoverableConnectionError('connection already closed')  
RecoverableConnectionError: connection already closed

It looks like, by the time nameko sends it's ack\_message, the worker's  
connection to rabbitmq is lost, so I get a RecoverableConnectionError. This  
means that the ack isn't transmitted correctly, so the message get's  
requeued, and the worker attempts to process the same message again.

Is there a way to avoid this? Right now I've done a very hacky fix: I  
dispatch events instead of making async\_calls. This ensures that the  
ack\_message occurs immediately, but I'd like to find a sustainable fix. Is  
it possible to set a timeout parameter (e.g. RpcProxy(timeout=20\*60) # 20  
min ) when I make the async\_call?

Thanks for all of your help! And sorry for the barrage of questions.

---

<div class="post-metadata">

**Author:** ![mattbennett](https://yyz1.discourse-cdn.com/flex031/user_avatar/discourse.nameko.io/mattbennett/32/13_2.png) [@mattbennett](https://discourse.nameko.io/u/mattbennett)\
**Post date:** [April 22, 2017, 10:20pm UTC](https://discourse.nameko.io/t/message-ack-failing-and-retriggering-task/191/2 "2017-04-22T22:20:07Z")

</div>

> Hi,
> 
> I have an rpc endpoint that does some processing and takes around 20 min  
> to complete. It basically kicks off a bunch of other async calls to other  
> rpc endpoints and waits for all of those to complete.  
> I frequently make async calls to this endpoint to do some processing.

However, I get this error message:

> Traceback (most recent call last):  
> &nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 177, in run  
> &nbsp;&nbsp;&nbsp;&nbsp;for \_ in self.consume(limit=None): # pragma: no cover  
> &nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 197, in consume  
> &nbsp;&nbsp;&nbsp;&nbsp;self.on\_iteration()  
> &nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 417, in on\_iteration  
> &nbsp;&nbsp;&nbsp;&nbsp;self.\_process\_pending\_message\_acks()  
> &nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 368, in  
> \_process\_pending\_message\_acks  
> &nbsp;&nbsp;&nbsp;&nbsp;msg.ack()  
> &nbsp;&nbsp;File "site-packages/kombu/message.py", line 88, in ack  
> &nbsp;&nbsp;&nbsp;&nbsp;self.channel.basic\_ack(self.delivery\_tag)  
> &nbsp;&nbsp;File "site-packages/amqp/channel.py", line 1584, in basic\_ack  
> &nbsp;&nbsp;&nbsp;&nbsp;self.\_send\_method((60, 80), args)  
> &nbsp;&nbsp;File "site-packages/amqp/abstract\_channel.py", line 50, in \_send\_method  
> &nbsp;&nbsp;&nbsp;&nbsp;raise RecoverableConnectionError('connection already closed')  
> RecoverableConnectionError: connection already closed
> 
> It looks like, by the time nameko sends it's ack\_message, the worker's  
> connection to rabbitmq is lost, so I get a RecoverableConnectionError. This  
> means that the ack isn't transmitted correctly, so the message get's  
> requeued, and the worker attempts to process the same message again.
> 
> Is there a way to avoid this? Right now I've done a very hacky fix: I  
> dispatch events instead of making async\_calls. This ensures that the  
> ack\_message occurs immediately, but I'd like to find a sustainable fix. Is  
> it possible to set a timeout parameter (e.g. RpcProxy(timeout=20\*60) # 20  
> min ) when I make the async\_call?

This isn't the RPC request timing out (there actually isn't a timeout on  
RPC requests, though it would be a useful feature). Rather it's the  
connection to Rabbit being closed. Are you connecting to Rabbit via a  
load-balancer or proxy of some kind? Oftentimes they will close idle  
connections after a number of minutes.

Which version of Nameko are you using? Until 2.5.4 message acks were sent  
by the consumer thread, and consumer heartbeats are enabled from 2.4.4  
onwards. If you're pre-2.4.4, upgrading will enable heartbeats that would  
stop the connection being determined as idle. In 2.5.4,  
RecoverableConnectionErrors are explicitly caught and retried (see  
[https://github.com/nameko/nameko/blob/v2.5.4/nameko/messaging.py#L326\](https://github.com/nameko/nameko/blob/v2.5.4/nameko/messaging.py#L326%5C)),  
although it's possible there's a bug there.

> Thanks for all of your help! And sorry for the barrage of questions.

No problem.

One other note -- rather than long-running RPC, have you considered using  
another entrypoint as a callback in the client, so it can be notified the  
work when it's done?

> **···**
>
> On Saturday, April 22, 2017 at 10:42:23 PM UTC+1, di...@clearmetal.com wrote:

---

<div class="post-metadata">

**Author:** ![Davis\_Kirkendall](https://yyz1.discourse-cdn.com/flex031/user_avatar/discourse.nameko.io/davis_kirkendall/32/18_2.png) [@Davis\_Kirkendall](https://discourse.nameko.io/u/Davis_Kirkendall)\
**Post date:** [September 25, 2017, 1:16pm UTC](https://discourse.nameko.io/t/message-ack-failing-and-retriggering-task/191/3 "2017-09-25T13:16:18Z")

</div>

I seem to be having the same Problem using nameko 2.6.0.  
Would there be a way of somehow reconnecting if a connection error es  
detected and retrying the ack instead of requeuing the job?

The reason for using the rpc jobs is so that the caller is blocked until  
the long running job returns (this is helpful for logging, timing and  
resource/cpu/memory management).

Thanks for the help and the great library!

> **···**
>
> Am Sonntag, 23. April 2017 00:20:08 UTC+2 schrieb Matt Yule-Bennett:
> 
> > On Saturday, April 22, 2017 at 10:42:23 PM UTC+1, di...@clearmetal.com \> wrote:
> > 
> > > Hi,
> > > 
> > > I have an rpc endpoint that does some processing and takes around 20 min  
> > > to complete. It basically kicks off a bunch of other async calls to other  
> > > rpc endpoints and waits for all of those to complete.  
> > > I frequently make async calls to this endpoint to do some processing.
> > 
> > However, I get this error message:
> > 
> > > Traceback (most recent call last):  
> > > &nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 177, in run  
> > > &nbsp;&nbsp;&nbsp;&nbsp;for \_ in self.consume(limit=None): # pragma: no cover  
> > > &nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 197, in consume  
> > > &nbsp;&nbsp;&nbsp;&nbsp;self.on\_iteration()  
> > > &nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 417, in on\_iteration  
> > > &nbsp;&nbsp;&nbsp;&nbsp;self.\_process\_pending\_message\_acks()  
> > > &nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 368, in  
> > > \_process\_pending\_message\_acks  
> > > &nbsp;&nbsp;&nbsp;&nbsp;msg.ack()  
> > > &nbsp;&nbsp;File "site-packages/kombu/message.py", line 88, in ack  
> > > &nbsp;&nbsp;&nbsp;&nbsp;self.channel.basic\_ack(self.delivery\_tag)  
> > > &nbsp;&nbsp;File "site-packages/amqp/channel.py", line 1584, in basic\_ack  
> > > &nbsp;&nbsp;&nbsp;&nbsp;self.\_send\_method((60, 80), args)  
> > > &nbsp;&nbsp;File "site-packages/amqp/abstract\_channel.py", line 50, in \_send\_method  
> > > &nbsp;&nbsp;&nbsp;&nbsp;raise RecoverableConnectionError('connection already closed')  
> > > RecoverableConnectionError: connection already closed
> > > 
> > > It looks like, by the time nameko sends it's ack\_message, the worker's  
> > > connection to rabbitmq is lost, so I get a RecoverableConnectionError. This  
> > > means that the ack isn't transmitted correctly, so the message get's  
> > > requeued, and the worker attempts to process the same message again.
> > > 
> > > Is there a way to avoid this? Right now I've done a very hacky fix: I  
> > > dispatch events instead of making async\_calls. This ensures that the  
> > > ack\_message occurs immediately, but I'd like to find a sustainable fix. Is  
> > > it possible to set a timeout parameter (e.g. RpcProxy(timeout=20\*60) # 20  
> > > min ) when I make the async\_call?
> > 
> > This isn't the RPC request timing out (there actually isn't a timeout on  
> > RPC requests, though it would be a useful feature). Rather it's the  
> > connection to Rabbit being closed. Are you connecting to Rabbit via a  
> > load-balancer or proxy of some kind? Oftentimes they will close idle  
> > connections after a number of minutes.
> > 
> > Which version of Nameko are you using? Until 2.5.4 message acks were sent  
> > by the consumer thread, and consumer heartbeats are enabled from 2.4.4  
> > onwards. If you're pre-2.4.4, upgrading will enable heartbeats that would  
> > stop the connection being determined as idle. In 2.5.4,  
> > RecoverableConnectionErrors are explicitly caught and retried (see  
> > [https://github.com/nameko/nameko/blob/v2.5.4/nameko/messaging.py#L326\](https://github.com/nameko/nameko/blob/v2.5.4/nameko/messaging.py#L326%5C)),  
> > although it's possible there's a bug there.
> > 
> > > Thanks for all of your help! And sorry for the barrage of questions.
> > 
> > No problem.
> > 
> > One other note -- rather than long-running RPC, have you considered using  
> > another entrypoint as a callback in the client, so it can be notified the  
> > work when it's done?

---

<div class="post-metadata">

**Author:** ![mattbennett](https://yyz1.discourse-cdn.com/flex031/user_avatar/discourse.nameko.io/mattbennett/32/13_2.png) [@mattbennett](https://discourse.nameko.io/u/mattbennett)\
**Post date:** [September 26, 2017, 12:08pm UTC](https://discourse.nameko.io/t/message-ack-failing-and-retriggering-task/191/4 "2017-09-26T12:08:59Z")

</div>

I think [RPC erroneously restarted if connection between worker and broker times out · Issue #477 · nameko/nameko · GitHub](https://github.com/nameko/nameko/issues/477) probably describes  
that's going on here. Do you have a workload that doesn't yield or block  
for I/O?

> **···**
>
> On Monday, September 25, 2017 at 2:16:18 PM UTC+1, Davis Kirkendall wrote:
> 
> > I seem to be having the same Problem using nameko 2.6.0.  
> > Would there be a way of somehow reconnecting if a connection error es  
> > detected and retrying the ack instead of requeuing the job?
> > 
> > The reason for using the rpc jobs is so that the caller is blocked until  
> > the long running job returns (this is helpful for logging, timing and  
> > resource/cpu/memory management).
> > 
> > Thanks for the help and the great library!
> > 
> > Am Sonntag, 23. April 2017 00:20:08 UTC+2 schrieb Matt Yule-Bennett:
> > 
> > > On Saturday, April 22, 2017 at 10:42:23 PM UTC+1, di...@clearmetal.com \>\> wrote:
> > > 
> > > > Hi,
> > > > 
> > > > I have an rpc endpoint that does some processing and takes around 20 min  
> > > > to complete. It basically kicks off a bunch of other async calls to other  
> > > > rpc endpoints and waits for all of those to complete.  
> > > > I frequently make async calls to this endpoint to do some processing.
> > > 
> > > However, I get this error message:
> > > 
> > > > Traceback (most recent call last):  
> > > > &nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 177, in run  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;for \_ in self.consume(limit=None): # pragma: no cover  
> > > > &nbsp;&nbsp;File "site-packages/kombu/mixins.py", line 197, in consume  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;self.on\_iteration()  
> > > > &nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 417, in on\_iteration  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;self.\_process\_pending\_message\_acks()  
> > > > &nbsp;&nbsp;File "site-packages/nameko/messaging.py", line 368, in  
> > > > \_process\_pending\_message\_acks  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;msg.ack()  
> > > > &nbsp;&nbsp;File "site-packages/kombu/message.py", line 88, in ack  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;self.channel.basic\_ack(self.delivery\_tag)  
> > > > &nbsp;&nbsp;File "site-packages/amqp/channel.py", line 1584, in basic\_ack  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;self.\_send\_method((60, 80), args)  
> > > > &nbsp;&nbsp;File "site-packages/amqp/abstract\_channel.py", line 50, in \_send\_method  
> > > > &nbsp;&nbsp;&nbsp;&nbsp;raise RecoverableConnectionError('connection already closed')  
> > > > RecoverableConnectionError: connection already closed
> > > > 
> > > > It looks like, by the time nameko sends it's ack\_message, the worker's  
> > > > connection to rabbitmq is lost, so I get a RecoverableConnectionError. This  
> > > > means that the ack isn't transmitted correctly, so the message get's  
> > > > requeued, and the worker attempts to process the same message again.
> > > > 
> > > > Is there a way to avoid this? Right now I've done a very hacky fix: I  
> > > > dispatch events instead of making async\_calls. This ensures that the  
> > > > ack\_message occurs immediately, but I'd like to find a sustainable fix. Is  
> > > > it possible to set a timeout parameter (e.g. RpcProxy(timeout=20\*60) # 20  
> > > > min ) when I make the async\_call?
> > > 
> > > This isn't the RPC request timing out (there actually isn't a timeout on  
> > > RPC requests, though it would be a useful feature). Rather it's the  
> > > connection to Rabbit being closed. Are you connecting to Rabbit via a  
> > > load-balancer or proxy of some kind? Oftentimes they will close idle  
> > > connections after a number of minutes.
> > > 
> > > Which version of Nameko are you using? Until 2.5.4 message acks were sent  
> > > by the consumer thread, and consumer heartbeats are enabled from 2.4.4  
> > > onwards. If you're pre-2.4.4, upgrading will enable heartbeats that would  
> > > stop the connection being determined as idle. In 2.5.4,  
> > > RecoverableConnectionErrors are explicitly caught and retried (see  
> > > [https://github.com/nameko/nameko/blob/v2.5.4/nameko/messaging.py#L326\](https://github.com/nameko/nameko/blob/v2.5.4/nameko/messaging.py#L326%5C)),  
> > > although it's possible there's a bug there.
> > > 
> > > > Thanks for all of your help! And sorry for the barrage of questions.
> > > 
> > > No problem.
> > > 
> > > One other note -- rather than long-running RPC, have you considered using  
> > > another entrypoint as a callback in the client, so it can be notified the  
> > > work when it's done?