""" Adversarial Challenge: Chat Concurrency, 64-Shard Routing, and Memory Bounding. Validates: multi-channel load across 64 shards, fan-out latency p99 < 15ms, 100-msg ring buffer rollover, duplicate spam prevention, whisper blacklist, and shard entropy. """ from __future__ import annotations import asyncio import time import tracemalloc from typing import List import pytest from server.chat.chat_service import ChatService from server.chat.chat_types import ChatChannelType, ChatMessageDTO, SendChatRequestDTO @pytest.fixture def chat_service() -> ChatService: return ChatService(cluster_shards=64) class TestChatConcurrencyAndRoutingAdversarial: """Adversarial stress and concurrency challenge harness for Milestone M1.""" @pytest.mark.asyncio async def test_concurrent_multi_channel_load_64_shards(self, chat_service: ChatService) -> None: """Challenge H1: 80 concurrent async publishers across all 8 channels and 64 shards.""" received_messages: List[ChatMessageDTO] = [] channels = [ "world", "recruit", "system", "zone:ancient_barrow", "zone:iron_forge", "guild:iron_clan", "party:raid_team_1", "feedback:9001", ] for ch in channels: for sub_id in range(1, 21): chat_service.subscribe_client(ch, sub_id, lambda msg: received_messages.append(msg)) # Launch 80 concurrent publishers across channels async def publisher(idx: int) -> bool: ch_type = list(ChatChannelType)[idx % len(ChatChannelType)] req = SendChatRequestDTO( sender_id=200_000 + idx, sender_name=f"Tester_{idx}", sender_level=50, channel=ch_type, content=f"Concurrent message payload {idx}", zone_id="ancient_barrow", guild_id="iron_clan", party_id="raid_team_1", target_id=200_001 if ch_type == ChatChannelType.WHISPER else 0, ) res = await chat_service.handle_send_chat(req) return res.success tasks = [publisher(i) for i in range(80)] results = await asyncio.gather(*tasks, return_exceptions=False) assert all(results) is True assert len(received_messages) > 0 # Verify active distribution in router shards non_empty_shards = sum( 1 for shard in chat_service.cluster_router.registry.shards if len(shard) > 0 ) assert non_empty_shards >= 5 @pytest.mark.asyncio async def test_fan_out_latency_percentiles_burst_sla_15ms(self, chat_service: ChatService) -> None: """Challenge H2: Fan-out latency percentiles under burst load meet p99 < 15ms SLA.""" delivery_count = 0 def sync_sink(msg: ChatMessageDTO) -> None: nonlocal delivery_count delivery_count += 1 async def async_sink(msg: ChatMessageDTO) -> None: nonlocal delivery_count delivery_count += 1 # Phase A: 1,000 mixed sync/async listeners on World channel (matching 1k edge benchmark) for i in range(1, 1001): cb = sync_sink if i % 10 != 0 else async_sink # 10% async coroutines chat_service.subscribe_client("world", i, cb) for b in range(100): req = SendChatRequestDTO( sender_id=300_000 + b, sender_name=f"Broadcaster_{b}", sender_level=40, channel=ChatChannelType.WORLD, content=f"World broadcast pulse #{b}", ) res = await chat_service.handle_send_chat(req) assert res.success is True assert delivery_count == 100 * 1000 stats = chat_service.cluster_router.get_latency_stats() assert stats["p99"] < 15.0, f"Mixed p99 latency {stats['p99']}ms breached 15.0ms SLA!" # Phase B: 4,000 high-density synchronous buffer sinks (Websocket ring buffers) chat_service_sync = ChatService(cluster_shards=64) sync_deliveries = 0 def mem_sink(msg: ChatMessageDTO) -> None: nonlocal sync_deliveries sync_deliveries += 1 for i in range(1, 4001): chat_service_sync.subscribe_client("world", i, mem_sink) for b in range(50): req = SendChatRequestDTO( sender_id=350_000 + b, sender_name=f"SyncBroadcaster_{b}", sender_level=40, channel=ChatChannelType.WORLD, content=f"Sync pulse #{b}", ) res = await chat_service_sync.handle_send_chat(req) assert res.success is True assert sync_deliveries == 50 * 4000 sync_stats = chat_service_sync.cluster_router.get_latency_stats() assert sync_stats["p99"] < 15.0, f"Sync p99 latency {sync_stats['p99']}ms breached 15.0ms SLA!" @pytest.mark.asyncio async def test_ring_buffer_memory_bounding_and_rollover(self, chat_service: ChatService) -> None: """Challenge H3: 100-message ring buffer strictly bounds memory under 1,500 messages.""" tracemalloc.start() mem_start, _ = tracemalloc.get_traced_memory() channel_key = "zone:ancient_barrow" # Ingest 1,500 messages through the channel manager for i in range(1500): msg = ChatMessageDTO( message_id=f"msg_{i}", sender_id=400_000 + i, sender_name=f"Player_{i}", sender_title="", sender_guild_tag="", channel=ChatChannelType.ZONE, target_id=0, raw_content=f"Barrow combat callout #{i}", filtered_content=f"Barrow combat callout #{i}", timestamp_ms=int(time.time() * 1000), is_censored=False, moderation_risk_score=0, linked_items=(), zone_id="ancient_barrow", guild_id="", party_id="", ) chat_service.channel_manager.add_to_history(channel_key, msg) history = chat_service.get_channel_history(channel_key) assert len(history) == 100, f"History length {len(history)} exceeded 100!" # Strict FIFO check: oldest remaining message must be #1400, newest #1499 assert history[0].message_id == "msg_1400" assert history[-1].message_id == "msg_1499" mem_end, _ = tracemalloc.get_traced_memory() tracemalloc.stop() delta_mb = (mem_end - mem_start) / (1024 * 1024) assert delta_mb < 2.0, f"Memory delta {delta_mb:.2f}MB indicates unbounded buffer growth!" @pytest.mark.asyncio async def test_duplicate_spam_prevention_world_and_zone(self, chat_service: ChatService) -> None: """Challenge H4: Duplicate spam within 5s is rejected on World/Zone, allowed on Party.""" # 1. World Channel duplicate detection req1 = SendChatRequestDTO( sender_id=500_001, sender_name="SpamCandidate", sender_level=25, channel=ChatChannelType.WORLD, content="WTS Than Ma Kiem +15 gia cuc re!", ) res1 = await chat_service.handle_send_chat(req1) assert res1.success is True # Rapid duplicate send from same player with whitespace and uppercase variation req1_dup = SendChatRequestDTO( sender_id=500_001, sender_name="SpamCandidate", sender_level=25, channel=ChatChannelType.WORLD, content=" wts than ma kiem +15 gia cuc re! ", ) res1_dup = await chat_service.handle_send_chat(req1_dup) assert res1_dup.success is False # Either duplicate spam or token bucket cooldown is triggered assert "spam" in res1_dup.error_message.lower() or "nhanh" in res1_dup.error_message.lower() # 2. Party channel should NOT trigger duplicate spam restriction req_party_1 = SendChatRequestDTO( sender_id=500_002, sender_name="Healer", sender_level=25, channel=ChatChannelType.PARTY, party_id="team_dungeon", content="Boss dang cast ultimate! Ne chieu!", ) res_p1 = await chat_service.handle_send_chat(req_party_1) assert res_p1.success is True # Advance time by 0.3s to pass party cooldown (0.2s) await asyncio.sleep(0.25) req_party_2 = SendChatRequestDTO( sender_id=500_002, sender_name="Healer", sender_level=25, channel=ChatChannelType.PARTY, party_id="team_dungeon", content="Boss dang cast ultimate! Ne chieu!", ) res_p2 = await chat_service.handle_send_chat(req_party_2) assert res_p2.success is True @pytest.mark.asyncio async def test_whisper_blacklist_and_protection_concurrency(self, chat_service: ChatService) -> None: """Challenge H5: Whisper protection (self-whisper, invalid target, blacklist under concurrency).""" # Self whisper rejection self_req = SendChatRequestDTO( sender_id=600_001, sender_name="Narcissus", sender_level=20, channel=ChatChannelType.WHISPER, target_id=600_001, content="Talking to myself", ) self_res = await chat_service.handle_send_chat(self_req) assert self_res.success is False assert "chính mình" in self_res.error_message # Invalid target rejection inv_req = SendChatRequestDTO( sender_id=600_001, sender_name="Narcissus", sender_level=20, channel=ChatChannelType.WHISPER, target_id=0, content="Talking to void", ) inv_res = await chat_service.handle_send_chat(inv_req) assert inv_res.success is False assert "không hợp lệ" in inv_res.error_message # Blacklist test: Target 600_002 blocks 600_001 chat_service.set_blacklist_entry(player_id=600_002, blocked_player_id=600_001, is_blocked=True) blocked_req = SendChatRequestDTO( sender_id=600_001, sender_name="Sender", sender_level=20, channel=ChatChannelType.WHISPER, target_id=600_002, content="Can you hear me?", ) blocked_res = await chat_service.handle_send_chat(blocked_req) assert blocked_res.success is False assert "đã chặn" in blocked_res.error_message # Concurrency stress: 30 concurrent whispers from non-blocked senders while toggling blacklist async def whisper_attempt(idx: int) -> bool: r = SendChatRequestDTO( sender_id=600_100 + idx, sender_name=f"Fan_{idx}", sender_level=20, channel=ChatChannelType.WHISPER, target_id=600_002, content=f"Hello fan mail #{idx}", ) resp = await chat_service.handle_send_chat(r) return resp.success tasks = [whisper_attempt(i) for i in range(30)] results = await asyncio.gather(*tasks) assert all(results) is True @pytest.mark.asyncio async def test_persistent_mute_cross_channel_isolation(self, chat_service: ChatService) -> None: """Challenge H6: RMT violation persists auto-mute across all 8 channels.""" sender_id = 700_001 rmt_req = SendChatRequestDTO( sender_id=sender_id, sender_name="RMTBot", sender_level=30, channel=ChatChannelType.WORLD, content="Ban vang SLL gia re lien he telegram @shopvang chuyen khoan atm momo 0912345678", ) rmt_res = await chat_service.handle_send_chat(rmt_req) assert rmt_res.success is False assert rmt_res.risk_score >= 60 # Verify persistent mute blocks sending on ALL channels test_channels = [ (ChatChannelType.WORLD, "", "", ""), (ChatChannelType.ZONE, "ancient_barrow", "", ""), (ChatChannelType.RECRUIT, "", "", ""), (ChatChannelType.GUILD, "", "clan_1", ""), (ChatChannelType.PARTY, "", "", "party_1"), (ChatChannelType.WHISPER, "", "", ""), (ChatChannelType.FEEDBACK, "", "", ""), ] for ch, z_id, g_id, p_id in test_channels: probe_req = SendChatRequestDTO( sender_id=sender_id, sender_name="RMTBot", sender_level=30, channel=ch, content="Trying to evade mute!", zone_id=z_id, guild_id=g_id, party_id=p_id, target_id=700_999 if ch == ChatChannelType.WHISPER else 0, ) probe_res = await chat_service.handle_send_chat(probe_req) assert probe_res.success is False assert "tạm khóa chat" in probe_res.error_message # Unmute and verify recovery chat_service.unmute_player(sender_id) is_muted, _ = chat_service.is_player_muted(sender_id) assert is_muted is False def test_shard_distribution_entropy(self, chat_service: ChatService) -> None: """Challenge H7: 1,000 distinct channels achieve balanced distribution across all 64 shards.""" registry = chat_service.cluster_router.registry sample_channels: List[str] = ["world", "recruit", "system"] for i in range(250): sample_channels.append(f"zone:dungeon_{i}") sample_channels.append(f"guild:syndicate_{i}") sample_channels.append(f"party:team_{i}") sample_channels.append(f"whisper:100{i}:200{i}") for ch in sample_channels: registry.subscribe(ch, client_id=1, callback=lambda msg: None) shard_occupancy = [len(shard) for shard in registry.shards] non_empty = sum(1 for count in shard_occupancy if count > 0) # Invariant: 64 shards should all be utilized with >= 1000 channels assert non_empty == 64, f"Only {non_empty}/64 shards populated! Distribution entropy too low." avg_per_shard = len(sample_channels) / 64.0 # No shard should hold more than 3.5x average max_in_shard = max(shard_occupancy) assert max_in_shard < avg_per_shard * 3.5, f"Extreme clustering in shard: {max_in_shard}"