diff --git a/remoulade/composition.py b/remoulade/composition.py index 6c70597c8..cc28b0d7b 100644 --- a/remoulade/composition.py +++ b/remoulade/composition.py @@ -41,6 +41,7 @@ class GroupInfo(namedtuple("GroupInfo", ("group_id", "children_count", "cancel_o group_id(str): The id of the group children_count(int) cancel_on_error(bool) + forget_on_enqueue """ def __new__(cls, *, group_id: str, children_count: int, cancel_on_error: bool): diff --git a/remoulade/middleware/pipelines.py b/remoulade/middleware/pipelines.py index c4ca935f1..dd4e0708f 100644 --- a/remoulade/middleware/pipelines.py +++ b/remoulade/middleware/pipelines.py @@ -94,8 +94,12 @@ def _send_next_message(self, pipe_target, broker, result, group_info): broker.enqueue(next_message) + if not pipe_ignore and group_info: + # We can forget the data now that it has been enqueued + self._group_results(group_info, broker, forget=True) + @staticmethod - def _group_results(group_info, broker): + def _group_results(group_info, broker, forget=False): """ Get the result of a group (fetch the group members message_ids then all the results) """ from ..collection_results import CollectionResults @@ -106,7 +110,7 @@ def _group_results(group_info, broker): message_ids = result_backend.get_group_message_ids(group_id=group_info.group_id) results = CollectionResults.from_message_ids(message_ids) - return list(results.get()) + return list(results.get(forget=forget)) def _group_completed(self, message, group_info, broker): """ Returns true if a group is completed, and increment the completion count of the group