Pubsub: let a queue subscribe to the same topic more than once

reboot/std/pubsub/v1/pubsub.py

@@ -72,9 +72,12 @@ async def Subscribe(
         context: WriterContext,
         request: SubscribeRequest,
     ) -> SubscribeResponse:
-        # Add subscriber to topic.
-        #
-        self.state.queue_ids.append(request.queue_id)
+        # Add subscriber to topic. Subscribing a queue that already
+        # subscribes to this topic is a no-op: a repeated queue id
+        # would make `Broker` call `Enqueue` on that queue twice using
+        # a single `context`, which Reboot refuses.
+        if request.queue_id not in self.state.queue_ids:
+            self.state.queue_ids.append(request.queue_id)
 
         # If this is a new topic, we'll need to schedule the broker.
         if not self.state.broker_started:
@@ -111,9 +114,14 @@ async def slice_items(
                 have_items,
             )
 
+            # In an earlier implementation there was a bug where a topic
+            # could be subscribed to by the same queue id more than
+            # once. That bug has been fixed, but to handle any `Queue`
+            # instances that have a repeated queue id already persisted
+            # we also deduplicate here via `dict.fromkeys`.
             await concurrently(
                 Queue.ref(queue_id).Enqueue(context, items=items)
-                for queue_id in queue_ids
+                for queue_id in dict.fromkeys(queue_ids)
             )
 
         return BrokerResponse()

tests/reboot/std/pubsub/v1/pubsub_tests.py

@@ -272,6 +272,45 @@ async def test_example_bulk_code_for_documentation(self) -> None:
         self.assertEqual(len(items.items), 5)
         self.assertEqual(as_str(items.items[2].value), "apple")
 
+    async def test_subscribe_twice(self) -> None:
+        """
+        Test that subscribing the same queue to a topic twice still
+        delivers every published item exactly once.
+        """
+        await self.rbt.up(
+            Application(
+                libraries=[
+                    pubsub_library(),
+                    queue_library(),
+                    sorted_map_library(),
+                ]
+            )
+        )
+
+        context = self.rbt.create_external_context(
+            name=f"test-{self.id()}",
+            app_internal=True,
+        )
+
+        test_topic = Topic.ref("test-topic")
+        test_queue = Queue.ref("receiving-queue")
+
+        # Subscribing the same queue twice leaves the topic with a
+        # single subscription for it, so the broker calls `Enqueue` on
+        # that queue exactly once per publish.
+        await test_topic.subscribe(context, queue_id=test_queue.state_id)
+        await test_topic.subscribe(context, queue_id=test_queue.state_id)
+
+        # Publish to the topic.
+        await test_topic.publish(context, bytes=b"first message")
+        await test_topic.publish(context, bytes=b"second message")
+
+        # Both messages arrive exactly once, in order.
+        message1 = await test_queue.dequeue(context)
+        message2 = await test_queue.dequeue(context)
+        self.assertEqual(message1.bytes, b"first message")
+        self.assertEqual(message2.bytes, b"second message")
+
 
 if __name__ == '__main__':
     unittest.main()