Repository: mercadona/rele

Issue:
Worker initialization should raise an error on duplicate subscriptions
If two `subs` subscribe to the same topic and don't declare a suffix, stealing will happen where approximately half the messages will be processed by one subscription and half by the other. That's extremely difficult to test against and to debug in production, so a protection would be very useful.

I'd suggest raising an error straight away so the worker doesn't start. A warning log wouldn't be as effective because it could go unnoticed.

Discussion:
EmilioCarrion: There were two options to implement this logic. In the worker inicialization or in the load of subscriptions from the modules. I decided to implement it in the load due to that we might want to instanciate multiple workers with the subscriptions spread between them and may be repeated but not in a single Worker.

Patch:
diff --git a/rele/config.py b/rele/config.py
index 67f97df..da3ae8e 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -87,7 +87,7 @@ def subscription_from_attribute(attribute):
 
 def load_subscriptions_from_paths(sub_module_paths, sub_prefix=None, filter_by=None):
 
-    subscriptions = []
+    subscriptions = {}
     for sub_module_path in sub_module_paths:
         sub_module = importlib.import_module(sub_module_path)
         for attr_name in dir(sub_module):
@@ -102,5 +102,13 @@ def load_subscriptions_from_paths(sub_module_paths, sub_prefix=None, filter_by=N
             if filter_by and not subscription.filter_by:
                 subscription.set_filters(filter_by)
 
-            subscriptions.append(subscription)
-    return subscriptions
+            if subscription.name in subscriptions:
+                found_subscription = subscriptions[subscription.name]
+                raise RuntimeError(
+                    f"Duplicate subscription name found: {subscription.name}. Subs "
+                    f"{subscription._func.__module__}.{subscription._func.__name__} and "
+                    f"{found_subscription._func.__module__}.{found_subscription._func.__name__} collide."
+                )
+
+            subscriptions[subscription.name] = subscription
+    return list(subscriptions.values())


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
index 00612f0..7a26be4 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -15,6 +15,11 @@ def sub_stub(data, **kwargs):
     return data["id"]
 
 
+@sub(topic="another-cool-topic", prefix="rele")
+def another_sub_stub(data, **kwargs):
+    return data["id"]
+
+
 class TestLoadSubscriptions:
     @pytest.fixture
     def subscriptions(self):
@@ -25,8 +30,8 @@ class TestLoadSubscriptions:
         )
 
     def test_load_subscriptions_in_a_module(self, subscriptions):
-        assert len(subscriptions) == 1
-        func_sub = subscriptions[0]
+        assert len(subscriptions) == 2
+        func_sub = subscriptions[-1]
         assert isinstance(func_sub, Subscription)
         assert func_sub.name == "rele-test-topic"
         assert func_sub({"id": 4}, lang="en") == 4
@@ -44,6 +49,16 @@ class TestLoadSubscriptions:
         assert klass_sub.name == "test-alternative-cool-topic"
         assert klass_sub({"id": 4}, lang="en") == 4
 
+    def test_raises_error_when_subscription_is_duplicated(self):
+        with pytest.raises(RuntimeError) as excinfo:
+            load_subscriptions_from_paths(["tests.test_config", "tests.more_subs.subs"])
+
+        assert (
+            str(excinfo.value)
+            == "Duplicate subscription name found: rele-another-cool-topic. Subs "
+            "tests.more_subs.subs.another_sub_stub and tests.test_config.another_sub_stub collide."
+        )
+
     def test_returns_sub_value_when_filtered_value_applied(self, subscriptions):
 
         assert subscriptions[-1]({"id": 4}, lang="en") == 4