Make sure AtMostOnceBarrier only releases once (or never) (#4)

I realized the implementation of `incr(self, key, amount, maximum, ttl)`
assumes a default value of `0` if the key does not exist. This means,
when the backend (e.g. Redis) loses its contents, we would assume every
`AtMostOnceBarrier` never ran. To fix this, the key is now created with
a value of `-1` (instead of `0`) and we only release if by incrementing
by `1` we stay under the maximum of `0` - which can only be true if the
previous value was `-1` and not with the default of `0`.

I also added some tests to confirm this bug existed and to confirm this
PR fixes it.

Author: pencil

dramatiq_workflow/_barrier.py

@@ -20,7 +20,7 @@ def __init__(self, backend, key, *args, ttl=900000):
         self.ran_key = f"{key}_ran"
 
     def create(self, parties):
-        self.backend.add(self.ran_key, 0, self.ttl)
+        self.backend.add(self.ran_key, -1, self.ttl)
         return super().create(parties)
 
     def wait(self, *args, block=True, timeout=None):
@@ -32,7 +32,7 @@ def wait(self, *args, block=True, timeout=None):
 
         released = super().wait(*args, block=False)
         if released:
-            never_released = self.backend.incr(self.ran_key, 1, 1, self.ttl)
+            never_released = self.backend.incr(self.ran_key, 1, 0, self.ttl)
             return never_released
 
         return False

dramatiq_workflow/tests/test_barrier.py

@@ -0,0 +1,39 @@
+import unittest
+
+from dramatiq.rate_limits.backends import StubBackend
+
+from .._barrier import AtMostOnceBarrier
+
+
+class AtMostOnceBarrierTests(unittest.TestCase):
+    def setUp(self):
+        self.backend = StubBackend()
+        self.key = "test_barrier"
+        self.parties = 3
+        self.ttl = 900000
+        self.barrier = AtMostOnceBarrier(self.backend, self.key, ttl=self.ttl)
+
+    def test_wait_block_true_raises(self):
+        with self.assertRaises(ValueError) as context:
+            self.barrier.wait(block=True)
+        self.assertEqual(str(context.exception), "Blocking is not supported by AtMostOnceBarrier")
+
+    def test_wait_releases_once(self):
+        self.barrier.create(self.parties)
+        for _ in range(self.parties - 1):
+            result = self.barrier.wait(block=False)
+            self.assertFalse(result)
+        result = self.barrier.wait(block=False)
+        self.assertTrue(result)
+        result = self.barrier.wait(block=False)
+        self.assertFalse(result)
+
+    def test_wait_does_not_release_when_db_emptied(self):
+        """
+        If the store is emptied, the barrier should not be released.
+        """
+        self.barrier.create(self.parties)
+        self.backend.db = {}
+        for _ in range(self.parties):
+            result = self.barrier.wait(block=False)
+            self.assertFalse(result)