Skipped messages can get inconsistent results stored

[FinnLidbetter]

If a message is dropped because of the Age Limit middleware, then getting the result of that message will be inconsistent depending on how soon after the message is dropped an attempt to get the result is made.

Not sure what the resolution should be here. I think it should either consistently store a None result, or consistently store an exception that is not Exception("unknown").

What OS are you using?

Ubuntu 20.04

What version of Dramatiq are you using?

1.11.0

What did you do?

Use a broker with the AgeLimit and Results middleware. Have a message hit its age limit and get its result in a blocking get_result call. Get the result for the same message again a second or two after the blocking get_result returns a result.

What did you expect would happen?

The result should not change.

What happened?

On the first get_result, the result is None.
On the second get_result, retrieving the result raises a ResultFailure exception with exception Exception("unknown")

This appears to be because:

1. When the Age Limit expires, the AgeLimit middleware sets the failed flag on the MessageProxy before the message is processed

   dramatiq/dramatiq/middleware/age_limit.py

   Lines 41 to 50 in e1f3245

   	def before_process_message(self, broker, message):
   	actor = broker.get_actor(message.actor_name)
   	max_age = message.options.get("max_age") or actor.options.get("max_age", self.max_age)
   	if not max_age:
   	return
   	if current_millis() - message.message_timestamp >= max_age:
   	self.logger.warning("Message %r has exceeded its age limit.", message.message_id)
   	message.fail()
   	return

2. In worker.py this causes and emit_after of "process_message" with a result of None

   dramatiq/dramatiq/worker.py

   Line 493 in e1f3245

   self.broker.emit_after("process_message", message, result=res)

3. The Results middleware's after_process_message will then store this None result and no exception

   dramatiq/dramatiq/results/middleware.py

   Line 95 in e1f3245

   self.backend.store_result(message, result, result_ttl)

4. After this happens, worker.py still has some post-processing of the message to do. Since the message was marked with failed, post_process_message will do an emit_after of "nack"

   dramatiq/dramatiq/worker.py

   Line 347 in e1f3245

   self.broker.emit_after("nack", message)

5. Since there was no exception, the results middleware will then replace the result at the message key in the result store with

   dramatiq/dramatiq/results/middleware.py

   Lines 105 to 109 in e1f3245

   	def after_nack(self, broker, message):
   	store_results, result_ttl = self._lookup_options(broker, message)
   	if store_results and message.failed:
   	exception = message._exception or Exception("unknown")
   	self.backend.store_exception(message, exception, result_ttl)

Sample Code

import dramatiq
import time

from dramatiq.brokers.rabbitmq import RabbitmqBroker
from dramatiq.results.backends import RedisBackend
from dramatiq.results import Results

result_backend = RedisBackend()
broker = RabbitmqBroker()
broker.add_middleware(Results(backend=result_backend))
dramatiq.set_broker(broker)

@dramatiq.actor(store_results=True, max_age=1)
def sample_actor():
  return 42

if __name__ == "__main__":
    message = sample_actor.send()
    result_1 = message.get_result(block=True, timeout=2000)
    print(result_1)
    time.sleep(2)
    # An Exception("unknown") will be raised here.
    result_2 = message.get_result()
    print(result_2)

[FinnLidbetter]

If an approach can be agreed upon, I would be happy to submit a PR to fix it.

It seems like the Age Limit middleware should at least be raising SkipMessage(), to trigger any after_skip_message hooks instead of after_process_message. But I'm not sure that this is sufficient to fix the problem. Since SkipMessage is caught in worker.py, if the message is still marked as failed, but has no exception, then I think it will still cause Exception("unknown") to be stored by the Results middleware when the message is nack'd due to message failure.

[Bogdanp]

Maybe AgeLimit should both raise SkipMessage and set an exception on the message? I think in that situation, you'll always end up with a consistent result (i.e. an exception) and that exception will at least describe the problem.

[FinnLidbetter]

Ok, what about raising SkipMessage in the AgeLimit middleware and then changing the except SkipMessage behaviour in _WorkerThread's process_message to stuff SkipMessage exceptions into the message, much like how BaseExceptions are stuffed into messages?

dramatiq/dramatiq/worker.py

Lines 495 to 504 in fa732f4

	except SkipMessage:
	self.logger.warning("Message %s was skipped.", message)
	self.broker.emit_after("skip_message", message)
	except BaseException as e:
	# Stuff the exception into the message [proxy] so that it
	# may be used by the stub broker to provide a nicer
	# testing experience. Also used by the results middleware
	# to pass exceptions into results.
	message.stuff_exception(e)

If that sounds good, I can submit a PR for this.

[Bogdanp]

Yes, I think that would be fine.

[FinnLidbetter]

Hi @Bogdanp, any thoughts on the PR I put up for this? #449

[Bogdanp]

Hey @FinnLidbetter, I took a look at it a while back, but I need to think about it some more because it might not be the right approach after all. I'll try to find some time to look at it again this weekend or the next.

[FinnLidbetter]

Ok no problem, that sounds good to me, thank you. Let me know if there's anything I can do to help.

[FinnLidbetter]

Hi @Bogdanp, sorry to be a bother on this again. Have you had time to give this one some thought? I am happy to take another crack at this if you can articulate your concern with the approach used in #449