Scaling Beyond One Process
An in-memory hub only reaches one process's clients, so a message broker between replicas is what makes two of them correct.
sub := rdb.Subscribe(ctx, "events"); for m := range sub.Channel() { hub.broadcast <- m.Payload }A hub lives inside one process. It knows the connections that process accepted and nothing else. Run two replicas of your service behind a load balancer and roughly half the clients attach to each, so a message published inside one replica reaches only its own half.
Sticky sessions do not fix this. Pinning a client to one replica keeps that client on one hub; it does nothing to carry a message to the other hub. The missing piece is a bus between the processes.
The shape that works
Each replica subscribes to one channel on a broker such as Redis, NATS or Kafka. A handler with news to share publishes it to the broker instead of writing to its own hub. Every replica, including the one that published, receives it from the broker and fans it out to its own clients. Adding replicas then changes nothing about correctness.
What the broker gives you
Redis pub/sub is the smallest version: fire and forget, with no replay. A replica that is restarting misses everything sent while it was down. A Redis stream, a Kafka topic or NATS JetStream keeps the messages, so a replica or a client can ask for everything after the last id it saw. That is what a feed needs when a gap would be noticed.
What trips people up
Long connections meet infrastructure built for short ones. An AWS load balancer closes an idle connection after 60 seconds by default, nginx does the same after 60 seconds of silence, and every rolling deploy drops every connection it holds. Clients must reconnect and resume, so put an id on each message and accept one from the client.
Example
package main
import (
"fmt"
"sync"
)
// broker stands in for Redis, NATS or Kafka: one Publish reaches every
// subscriber. Swapping this type for a real client is the only change needed
// to make the rest of this program work across machines.
type broker struct {
mu sync.Mutex
subs []chan string
}
func (b *broker) Subscribe() <-chan string {
b.mu.Lock()
defer b.mu.Unlock()
ch := make(chan string, 8)
b.subs = append(b.subs, ch)
return ch
}
func (b *broker) Publish(msg string) {
b.mu.Lock()
defer b.mu.Unlock()
for _, ch := range b.subs {
ch <- msg // buffered, so one slow replica cannot stall the publisher
}
}
// replica is one copy of the service. Its hub knows only the clients that
// connected to it, so it hears about everything else here.
type replica struct {
name string
in <-chan string
}
func main() {
b := &broker{}
replicas := []*replica{
{name: "pod-a", in: b.Subscribe()},
{name: "pod-b", in: b.Subscribe()},
}
// A request lands on pod-a. It publishes instead of writing to its own hub.
b.Publish("price: 42")
for _, r := range replicas {
fmt.Printf("%s fans out %q to its own clients\n", r.name, <-r.in)
}
}
// Output:
// pod-a fans out "price: 42" to its own clients
// pod-b fans out "price: 42" to its own clients
When to use it
- Make chat keep working after the deployment is scaled from one pod to three.
- Keep dashboards correct through a rolling deploy that replaces every pod.
- Let a client that reconnects ask for the messages it missed while offline.
More examples
Bridging a hub to Redis pub/sub
The broker stand-in from the main example replaced with a real one, using github.com/redis/go-redis/v9. The rule to hold on to is in the comment: handlers call Publish and never touch the hub, so there is exactly one path from an event to a client and no replica can double-deliver. Run returns on context cancellation, which makes it something you can start and stop alongside the HTTP server.
package realtime
import (
"context"
"github.com/redis/go-redis/v9"
)
type hub struct {
broadcast chan string
}
// bridge connects one replica to every other replica. A handler never writes
// to its own hub; it publishes, and the subscription below does the fan-out.
type bridge struct {
rdb *redis.Client
channel string
hub *hub
}
func newBridge(addr, channel string, h *hub) *bridge {
return &bridge{
rdb: redis.NewClient(&redis.Options{Addr: addr}),
channel: channel,
hub: h,
}
}
// Publish is what an HTTP handler calls after it changes something.
func (b *bridge) Publish(ctx context.Context, msg string) error {
return b.rdb.Publish(ctx, b.channel, msg).Err()
}
// Run delivers every published message, including this replica's own, to the
// local hub. One subscription per process.
func (b *bridge) Run(ctx context.Context) error {
sub := b.rdb.Subscribe(ctx, b.channel)
defer sub.Close()
incoming := sub.Channel() // one Go channel carrying every published message
for {
select {
case <-ctx.Done():
return ctx.Err()
case m, open := <-incoming:
if !open {
return nil
}
b.hub.broadcast <- m.Payload
}
}
}When a gap matters: a stream instead of pub/sub
Pub/sub forgets a message the moment it is delivered. A Redis stream keeps it, so a replica that was restarting, or a client that was on a train, can ask for everything after the last id it saw rather than silently missing it. MaxLen is there because this is a buffer and not a database. Reach for this when a missed message is a bug rather than a cosmetic gap.
package realtime
import (
"context"
"github.com/redis/go-redis/v9"
)
// A Redis stream keeps the messages, so a replica that was restarting can ask
// for everything after the last id it saw. Pub/sub cannot do that.
type streamBridge struct {
rdb *redis.Client
stream string
lastID string // "$" means "only what arrives from now on"
}
func (s *streamBridge) Append(ctx context.Context, body string) error {
return s.rdb.XAdd(ctx, &redis.XAddArgs{
Stream: s.stream,
MaxLen: 10000, // trim: this is a buffer, not a database
Values: map[string]any{"body": body},
}).Err()
}
// Read blocks until there is something new, then returns it with the id that
// the caller should hand back next time.
func (s *streamBridge) Read(ctx context.Context) ([]string, error) {
res, err := s.rdb.XRead(ctx, &redis.XReadArgs{
Streams: []string{s.stream, s.lastID},
Block: 0, // 0 means wait indefinitely
}).Result()
if err != nil {
return nil, err
}
var out []string
for _, st := range res {
for _, entry := range st.Messages {
s.lastID = entry.ID
if body, ok := entry.Values["body"].(string); ok {
out = append(out, body)
}
}
}
return out, nil
}The infrastructure between you and the client
Correct Go code still dies at 60 seconds if the proxy in front of it is left at its defaults. These annotations raise the nginx ingress idle timeouts and turn off response buffering, which is the setting that makes an SSE endpoint look like it emits nothing. An AWS load balancer has its own idle timeout, also 60 seconds by default, set on the load balancer rather than here.
# A long-lived connection meets infrastructure built for short ones.
# These annotations are the usual reason a stream dies after 60 seconds.
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: realtime
annotations:
# Default is 60 seconds of silence, then the proxy closes the connection.
nginx.ingress.kubernetes.io/proxy-read-timeout: "3600"
nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
# Buffering holds events until the buffer fills, which looks like a bug.
nginx.ingress.kubernetes.io/proxy-buffering: "off"
spec:
ingressClassName: nginx
rules:
- host: app.example.com
http:
paths:
- path: /events
pathType: Prefix
backend:
service:
name: realtime
port:
number: 8080
Discussion