Use sharded pub/sub

[callmedenchick]

Setup Cluster Valkey, bridge. In 2 terminals:

Terminal 1: sending messages:

/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000001&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000002&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000003&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000001&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000002&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000003&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000004&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000005&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000006&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000007&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000008&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000009&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}
/ # curl -X POST "http://172.20.0.30:8081/bridge/message?client_id=sender&to=0000000000000000
000000000000000000000000000000000000000000000010&ttl=300&topic=test" -d "message $(date +%s)"
{"message":"OK","statusCode":200}

Terminal 2: receiving messeges:

...
vent: heartbeat

event: message
id: 115572233409015125
data: {"from":"sender","message":"message 1763492331","trace_id":"019a9855-7e3e-7942-8f22-80cb46655597","connect_source":{"ip":""}}

event: message
id: 115572233839717718
data: {"from":"sender","message":"message 1763492337","trace_id":"019a9855-97ea-7eb6-b45d-f843fd639e68","connect_source":{"ip":""}}

event: heartbeat

event: message
id: 115572234261835095
data: {"from":"sender","message":"message 1763492344","trace_id":"019a9855-b113-7655-8cfa-54cfee8c33c1","connect_source":{"ip":""}}

event: heartbeat

event: message
id: 115572234664815960
data: {"from":"sender","message":"message 1763492350","trace_id":"019a9855-c918-70f6-a3a1-b45c9b4ff881","connect_source":{"ip":""}}

event: message
id: 115572235029654873
data: {"from":"sender","message":"message 1763492355","trace_id":"019a9855-ded7-7d5a-9c71-b67cc2901c44","connect_source":{"ip":""}}

event: message
id: 115572235271286106
data: {"from":"sender","message":"message 1763492359","trace_id":"019a9855-ed3e-776e-917d-7db10cb8f7b1","connect_source":{"ip":""}}

event: heartbeat

event: message
id: 115572235544112475
data: {"from":"sender","message":"message 1763492363","trace_id":"019a9855-fd81-73f6-a3bd-997c47ffa618","connect_source":{"ip":""}}

event: message
id: 115572235915570524
data: {"from":"sender","message":"message 1763492369","trace_id":"019a9856-13a5-78ff-9b67-9167d3584fa6","connect_source":{"ip":""}}

event: heartbeat

event: message
id: 115572236225949021
data: {"from":"sender","message":"message 1763492374","trace_id":"019a9856-2625-7b37-90e2-7adc021d959c","connect_source":{"ip":""}}

event: heartbeat

event: message
id: 115572236655734110
data: {"from":"sender","message":"message 1763492380","trace_id":"019a9856-3fc3-7da7-86d5-dc791361ca4c","connect_source":{"ip":""}}

event: heartbeat

All messages received.

[github-actions]

📊 Performance Metrics

Performance Metrics (memory storage)

• CPU: 0.66s (4 cores) • Goroutines: 7 • Threads: 7
• Memory: 15.6MB heap • 53.0MB RAM • 27.2MB total • 232350 allocs
• GC: 5 cycles (0.69ms avg)
• FDs: 8/65536 (0.0%)

Performance Metrics (postgres storage)

• CPU: 0.77s (4 cores) • Goroutines: 9 • Threads: 9
• Memory: 15.8MB heap • 41.7MB RAM • 25.1MB total • 248298 allocs
• GC: 5 cycles (0.75ms avg)
• FDs: 34/65536 (0.1%)

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (dnsmasq storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

[github-actions]

📊 Performance Metrics

Performance Metrics (memory storage)

• CPU: 0.70s (4 cores) • Goroutines: 7 • Threads: 7
• Memory: 12.7MB heap • 45.1MB RAM • 25.4MB total • 231603 allocs
• GC: 6 cycles (0.55ms avg)
• FDs: 8/65536 (0.0%)

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (nginx storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (dnsmasq storage)

❌ Bridge service not accessible after 30 seconds

[github-actions]

📊 Performance Metrics

Performance Metrics (postgres storage)

• CPU: 0.78s (4 cores) • Goroutines: 9 • Threads: 9
• Memory: 15.2MB heap • 43.9MB RAM • 26.4MB total • 250053 allocs
• GC: 6 cycles (1.57ms avg)
• FDs: 34/65536 (0.1%)

[Copilot]

Pull Request Overview

This PR migrates the Valkey storage implementation from regular Redis pub/sub to sharded pub/sub (SPUBLISH/SSUBSCRIBE). The change uses Redis hash tags {client} in channel names to ensure all client channels are routed to the same shard in a Redis cluster, which is a requirement for sharded pub/sub to work correctly.

• Updated channel naming from client: to {client}: format with hash tags
• Migrated pub/sub methods to use sharded variants (SPublish, SSubscribe, SUnsubscribe)
• Updated channel parsing logic to handle the new format

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[github-actions]

📊 Performance Metrics

Performance Metrics (memory storage)

• CPU: 0.67s (4 cores) • Goroutines: 7 • Threads: 7
• Memory: 19.7MB heap • 48.9MB RAM • 27.9MB total • 233715 allocs
• GC: 5 cycles (0.25ms avg)
• FDs: 8/65536 (0.0%)

Performance Metrics (postgres storage)

• CPU: 0.79s (4 cores) • Goroutines: 9 • Threads: 9
• Memory: 21.8MB heap • 47.2MB RAM • 27.2MB total • 251095 allocs
• GC: 5 cycles (0.38ms avg)
• FDs: 34/65536 (0.1%)

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (dnsmasq storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (nginx storage)

❌ Bridge service not accessible after 30 seconds

[github-actions]

📊 Performance Metrics

Performance Metrics (memory storage)

• CPU: 0.68s (4 cores) • Goroutines: 7 • Threads: 8
• Memory: 7.2MB heap • 31.5MB RAM • 24.2MB total • 233393 allocs
• GC: 7 cycles (0.11ms avg)
• FDs: 8/65536 (0.0%)

Performance Metrics (postgres storage)

• CPU: 0.81s (4 cores) • Goroutines: 9 • Threads: 8
• Memory: 12.0MB heap • 48.2MB RAM • 28.7MB total • 251756 allocs
• GC: 6 cycles (1.53ms avg)
• FDs: 34/65536 (0.1%)

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (dnsmasq storage)

❌ Bridge service not accessible after 30 seconds

[github-actions]

📊 Performance Metrics

Performance Metrics (memory storage)

• CPU: 0.70s (4 cores) • Goroutines: 7 • Threads: 7
• Memory: 20.0MB heap • 49.1MB RAM • 26.8MB total • 232262 allocs
• GC: 5 cycles (0.74ms avg)
• FDs: 8/65536 (0.0%)

Performance Metrics (postgres storage)

• CPU: 0.80s (4 cores) • Goroutines: 9 • Threads: 11
• Memory: 13.4MB heap • 38.0MB RAM • 24.4MB total • 248892 allocs
• GC: 6 cycles (0.68ms avg)
• FDs: 34/65536 (0.1%)

📊 Performance Metrics (dnsmasq storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (cluster-valkey storage)

❌ Bridge service not accessible after 30 seconds

📊 Performance Metrics (nginx storage)

❌ Bridge service not accessible after 30 seconds