feat(pipeline): forget result of group pipelines

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):

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