diff --git a/reboot/std/pubsub/v1/pubsub.py b/reboot/std/pubsub/v1/pubsub.py index 21f451ae4..a40f86d94 100644 --- a/reboot/std/pubsub/v1/pubsub.py +++ b/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() diff --git a/tests/reboot/std/pubsub/v1/pubsub_tests.py b/tests/reboot/std/pubsub/v1/pubsub_tests.py index 9530f2d03..4e4a5c54f 100644 --- a/tests/reboot/std/pubsub/v1/pubsub_tests.py +++ b/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()