From e669e20e9c35268ab37b58cb3ecf1a0f3a2c3984 Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Fri, 11 Sep 2026 20:29:48 +0000 Subject: [PATCH 1/2] Let a queue subscribe to the same topic more than once `Topic.Subscribe` appended to `state.queue_ids` unconditionally, so subscribing the same queue to the same topic twice left a duplicate entry. `Topic.Broker` fans out over those ids, so a duplicate made it call `Enqueue` on one queue twice using a single `context`, which Reboot refuses: ValueError: To call 'rbt.std.collections.queue.v1.QueueMethods.Enqueue' of 'receiving-queue' more than once using the same context an idempotency alias or key must be specified The broker then retried with backoff forever and the topic never drained. That only happened when a second `Subscribe` landed before the broker had drained the topic, so it surfaced as a timeout under load: `//tests/reboot/std/pubsub/v1:test_pubsub_tests_ts` hit its 300s deadline in https://github.com/reboot-dev/reboot/actions/runs/34603229734/job/103275784303 `Subscribe` now skips a queue id the topic already carries, and `Broker` deduplicates the ids before fanning out, so a topic whose state accumulated duplicates before this fix drains instead of retrying forever. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01TQxdTEi8qsBfxCcuLxrNmQ --- reboot/std/pubsub/v1/pubsub.py | 16 ++++++--- tests/reboot/std/pubsub/v1/pubsub_tests.py | 39 ++++++++++++++++++++++ 2 files changed, 51 insertions(+), 4 deletions(-) diff --git a/reboot/std/pubsub/v1/pubsub.py b/reboot/std/pubsub/v1/pubsub.py index 21f451ae4..47da81c03 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, ) + # A topic subscribed to before `Subscribe` deduplicated may + # have a repeated queue id persisted. `until` memoizes the + # sliced list and `slice_items` has already taken those + # items out of state, so the duplicate replays on every + # retry; deduplicating here is what lets such a topic drain. 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() From 80bc24d7987d8a46fca22e74f4c9f7e15c97721e Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Mon, 14 Sep 2026 11:24:36 +0200 Subject: [PATCH 2/2] Apply review suggestion Co-authored-by: Benjamin Hindman --- reboot/std/pubsub/v1/pubsub.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/reboot/std/pubsub/v1/pubsub.py b/reboot/std/pubsub/v1/pubsub.py index 47da81c03..a40f86d94 100644 --- a/reboot/std/pubsub/v1/pubsub.py +++ b/reboot/std/pubsub/v1/pubsub.py @@ -114,11 +114,11 @@ async def slice_items( have_items, ) - # A topic subscribed to before `Subscribe` deduplicated may - # have a repeated queue id persisted. `until` memoizes the - # sliced list and `slice_items` has already taken those - # items out of state, so the duplicate replays on every - # retry; deduplicating here is what lets such a topic drain. + # 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 dict.fromkeys(queue_ids)