test: Stabilize sharded Redis broadcast multi-subscriber test (#29733)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
-LAN- 2025-12-16 16:35:55 +08:00 committed by GitHub
parent 4589157963
commit 0749e6e090
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 21 additions and 3 deletions

View File

@ -113,16 +113,31 @@ class TestShardedRedisBroadcastChannelIntegration:
topic = broadcast_channel.topic(topic_name)
producer = topic.as_producer()
subscriptions = [topic.subscribe() for _ in range(subscriber_count)]
ready_events = [threading.Event() for _ in range(subscriber_count)]
def producer_thread():
time.sleep(0.2) # Allow all subscribers to connect
deadline = time.time() + 5.0
for ev in ready_events:
remaining = deadline - time.time()
if remaining <= 0:
break
if not ev.wait(timeout=max(0.0, remaining)):
pytest.fail("subscriber did not become ready before publish deadline")
producer.publish(message)
time.sleep(0.2)
for sub in subscriptions:
sub.close()
def consumer_thread(subscription: Subscription) -> list[bytes]:
def consumer_thread(subscription: Subscription, ready_event: threading.Event) -> list[bytes]:
received_msgs = []
# Prime subscription so the underlying Pub/Sub listener thread starts before publishing
try:
_ = subscription.receive(0.01)
except SubscriptionClosedError:
return received_msgs
finally:
ready_event.set()
while True:
try:
msg = subscription.receive(0.1)
@ -137,7 +152,10 @@ class TestShardedRedisBroadcastChannelIntegration:
with ThreadPoolExecutor(max_workers=subscriber_count + 1) as executor:
producer_future = executor.submit(producer_thread)
consumer_futures = [executor.submit(consumer_thread, subscription) for subscription in subscriptions]
consumer_futures = [
executor.submit(consumer_thread, subscription, ready_events[idx])
for idx, subscription in enumerate(subscriptions)
]
producer_future.result(timeout=10.0)
msgs_by_consumers = []