Bugfix: Change logic of completed_count property for group (#814)

* debug completed_count property for group
* Add test for group.completion_count - Add test for for when the jobs in a group finish out of order.

---------

Co-authored-by: LincolnPuzey <[email protected]>

Author: ABolouk

CONTRIBUTORS.md

@@ -76,3 +76,4 @@ of those changes to CLEARTYPE SRL.
 | [@ksoviero-zengrc](https://github.com/ksoviero-zengrc) | Kevin Soviero          |
 | [@mikeroll](https://github.com/mikeroll)               | Mikhail Bulash         |
 | [@janek-cosmose](https://github.com/janek-cosmose)     | Jan Szejko             |
+| [@ABolouk](https://github.com/ABolouk) | Amirhossein Bolouk Asli |

dramatiq/composition.py

@@ -242,14 +242,16 @@ def completed_count(self):
         Returns:
           int: The total number of results.
         """
-        for count, child in enumerate(self.children, start=1):
+        count = 0
+        for child in self.children:
             try:
                 if isinstance(child, group):
                     child.get_results()
                 else:
                     child.get_result()
+                count += 1
             except ResultMissing:
-                return count - 1
+                pass
 
         return count
 

tests/test_composition.py

@@ -268,6 +268,38 @@ def wait(n):
     assert g.completed
 
 
+def test_groups_completion_count_works_for_tasks_finishing_out_of_order(stub_broker, stub_worker, result_backend):
+    # Regression test for https://github.com/Bogdanp/dramatiq/issues/452
+
+    # Given that I have a result backend
+    stub_broker.add_middleware(Results(backend=result_backend))
+
+    # And an actor that waits some amount of time, before notifying the Condition
+    condition = Condition()
+
+    @dramatiq.actor(store_results=True)
+    def wait(n):
+        time.sleep(n)
+        with condition:
+            condition.notify_all()
+            return n
+
+    # When I group messages of varying durations together and run the group.
+    # NOTE the durations ensure the jobs finish in a different order to what they are in the group.
+    g = group(wait.message(t) for t in [3.5, 1, 2.2])
+    g.run()
+
+    # Then every time a job in the group completes, the completed_count should increase by 1
+    for count in [1, 2, 3]:
+        with condition:
+            condition.wait(5)
+            time.sleep(0.1)  # give the worker time to set the result
+            assert g.completed_count == count
+
+    # Finally, completed should be true
+    assert g.completed
+
+
 def test_pipeline_does_not_continue_to_next_actor_when_message_is_marked_as_failed(stub_broker, stub_worker):
     # Given that I have an actor that fails messages
     class FailMessageMiddleware(middleware.Middleware):