WebSockets in Go With a Real Hub Design and Backpressure

A WebSocket looks easy until the second user connects. One connection is a loop that reads and writes. Two connections that talk to each other need somewhere for messages to go, and the moment that somewhere exists you have decisions about who may write when, what happens when a client stops reading, and who cleans up when one disappears. Most of the WebSocket bugs I have seen in Go services are not protocol bugs. They are a missing queue, a missing deadline or a map touched from two goroutines.

This article builds the structure that holds up: a hub, per client queues, deadlines in both directions, and an explicit policy for slow consumers. It is a project in the web cluster that ends in a production HTTP server, and it assumes you know why you want a WebSocket instead of server sent events or long polling. The concurrency ideas come from worker pools and channel ownership. Everything ran on Go 1.25.6 with github.com/gorilla/websocket v1.5.3.

Why gorilla, and what the standard library lacks

The standard library has no WebSocket package, although the protocol itself, RFC 6455, is short enough to read in an evening. The golang.org/x/net/websocket package exists but its own documentation says it lacks features and points to other libraries. In practice you choose between github.com/gorilla/websocket, the long-standing one with a huge amount of production use, and github.com/coder/websocket (formerly nhooyr), which is context aware and has a smaller API. I use gorilla here because its two rules, one reader and one writer at a time, are explicit and teach the right structure. The design in this article ports to either.

The package documentation states the concurrency contract in a single sentence: connections support one concurrent reader and one concurrent writer. Applications must make sure no two goroutines call write methods at the same time, and the same for read methods. The only exception is that Close and WriteControl may be called concurrently with the others. Everything about the design below follows from that sentence.

The shape of the system

Three pieces:

• A Client wraps one connection. It owns two goroutines, a readPump and a writePump, and a send channel with a small fixed capacity.
• A Hub holds the set of connected clients behind a mutex and exposes Add, Remove and Broadcast.
• An HTTP handler upgrades the request, builds a client, registers it and starts the pumps.

The classic example in gorilla's repository runs the hub as its own goroutine with register, unregister and broadcast channels. That works, and a mutex around a map is just as correct and a lot easier to read and test. What matters is not which you pick, but that Broadcast never blocks on a slow client.

The code

package main

import "sync"

// Client is one connection's outbound queue. The connection itself lives in conn.go.
type Client struct {
	id   int
	send chan []byte
}

// Hub keeps the set of clients and fans messages out to them.
type Hub struct {
	mu      sync.Mutex
	clients map[*Client]struct{}
	dropped int
}

func NewHub() *Hub { return &Hub{clients: make(map[*Client]struct{})} }

func (h *Hub) Add(c *Client) {
	h.mu.Lock()
	h.clients[c] = struct{}{}
	h.mu.Unlock()
}

func (h *Hub) Remove(c *Client) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if _, ok := h.clients[c]; ok {
		delete(h.clients, c)
		close(c.send) // the writer sees the closed channel and sends a close frame
	}
}

// Broadcast never blocks. A client whose queue is full is too slow, and is dropped.
func (h *Hub) Broadcast(msg []byte) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for c := range h.clients {
		select {
		case c.send <- msg:
		default:
			delete(h.clients, c)
			close(c.send)
			h.dropped++
		}
	}
}

func (h *Hub) Stats() (clients, dropped int) {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.clients), h.dropped
}

The important line is the default branch in Broadcast. A blocking send, c.send <- msg, would let one client that stopped reading freeze the loop, and with it every other client, because the mutex is held. A queue that is full means the peer has not kept up for several messages. For chat, presence and live dashboards the right policy is to disconnect it and let the browser reconnect and resync. For data where every message matters (an order feed), the policy is different: you need a replayable log and Last-Event-ID style resume rather than a bigger queue.

Closing c.send from the hub is safe because only the hub sends on it and it does so under the lock, which is the ownership rule from the channels article: exactly one party closes, and it is the one that sends.

Next, the connection side.

package main

import (
	"log"
	"net/http"
	"time"

	"github.com/gorilla/websocket"
)

const (
	writeWait  = 2 * time.Second
	pongWait   = 4 * time.Second
	pingPeriod = pongWait * 9 / 10
	maxMessage = 512
	queueSize  = 4
)

func newUpgrader(allowed ...string) websocket.Upgrader {
	return websocket.Upgrader{
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		CheckOrigin: func(r *http.Request) bool {
			origin := r.Header.Get("Origin")
			if origin == "" {
				return true // not a browser; a browser always sends Origin on a WebSocket handshake
			}
			for _, a := range allowed {
				if origin == a {
					return true
				}
			}
			return false
		},
	}
}

func serveWS(hub *Hub, up websocket.Upgrader, next func() int) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		conn, err := up.Upgrade(w, r, nil)
		if err != nil {
			return // Upgrade has already written the HTTP error
		}
		c := &Client{id: next(), send: make(chan []byte, queueSize)}
		hub.Add(c)
		go writePump(conn, c)
		readPump(conn, c, hub) // the handler goroutine is the reader
	}
}

// readPump is the only reader. It ends when the peer closes, errors, or misses its pongs.
func readPump(conn *websocket.Conn, c *Client, hub *Hub) {
	defer func() {
		hub.Remove(c)
		conn.Close()
	}()
	conn.SetReadLimit(maxMessage)
	conn.SetReadDeadline(time.Now().Add(pongWait))
	conn.SetPongHandler(func(string) error {
		return conn.SetReadDeadline(time.Now().Add(pongWait))
	})
	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
				log.Printf("client %d: %v", c.id, err)
			}
			return
		}
		hub.Broadcast(msg)
	}
}

// writePump is the only writer. It also sends the pings.
func writePump(conn *websocket.Conn, c *Client) {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		conn.Close()
	}()
	for {
		select {
		case msg, ok := <-c.send:
			conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
				return
			}
			if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil {
				return
			}
		case <-ticker.C:
			conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

Notice who does what. The handler goroutine becomes the reader, so there is no extra goroutine for it and the connection's lifetime equals the handler's. The writer is started separately and is the only code that writes, which satisfies the library's contract without a lock. When the hub closes c.send, the writer sends a proper close frame and returns, and its deferred conn.Close() unblocks the reader, whose own defer calls hub.Remove (a no-op the second time, thanks to the check inside it).

Four numbers in the constants deserve a comment. pongWait is how long the server tolerates silence: any frame from the peer is not enough, only a read that completes resets the deadline, and the pong handler does that for control frames. pingPeriod must be shorter than pongWait so a healthy client always has a pong land in time; nine tenths is the convention from the library's example. maxMessage bounds what ReadMessage allocates for one client. And queueSize is the backlog you will tolerate, small on purpose.

Trying it

The program below starts the server on httptest, connects three clients and exercises the rules: fan-out, a bad origin, an oversized message and a stuck consumer.

package main

import (
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"time"

	"github.com/gorilla/websocket"
)

func main() {
	hub := NewHub()
	var ids atomic.Int64
	next := func() int { return int(ids.Add(1)) }
	up := newUpgrader("https://app.example")

	srv := httptest.NewServer(serveWS(hub, up, next))
	defer srv.Close()
	url := "ws" + strings.TrimPrefix(srv.URL, "http")

	dial := func(origin string) (*websocket.Conn, *http.Response, error) {
		h := http.Header{}
		if origin != "" {
			h.Set("Origin", origin)
		}
		return websocket.DefaultDialer.Dial(url, h)
	}

	// 1. Fan-out: three clients, one speaks, all three hear it.
	var conns []*websocket.Conn
	for i := 0; i < 3; i++ {
		c, _, err := dial("https://app.example")
		if err != nil {
			panic(err)
		}
		conns = append(conns, c)
	}
	waitFor(func() bool { n, _ := hub.Stats(); return n == 3 })
	conns[0].WriteMessage(websocket.TextMessage, []byte("hello everyone"))
	for i, c := range conns {
		c.SetReadDeadline(time.Now().Add(2 * time.Second))
		_, msg, err := c.ReadMessage()
		fmt.Printf("client %d heard %q err=%v\n", i+1, msg, err)
	}

	// 2. A page on another origin is refused during the handshake.
	_, resp, err := dial("https://evil.example")
	fmt.Printf("bad origin: status=%d err=%v\n", resp.StatusCode, err)

	// 3. A message over the read limit closes the connection with 1009.
	conns[1].WriteMessage(websocket.TextMessage, []byte(strings.Repeat("x", 2000)))
	conns[1].SetReadDeadline(time.Now().Add(2 * time.Second))
	_, _, err = conns[1].ReadMessage()
	fmt.Printf("oversize message: %v\n", err)
	waitFor(func() bool { n, _ := hub.Stats(); return n == 2 })

	// 4. A consumer that never drains its queue is dropped on the fifth message (the queue holds four).
	stuck := &Client{id: 99, send: make(chan []byte, queueSize)}
	hub.Add(stuck)
	for i := 1; i <= 6; i++ {
		hub.Broadcast([]byte(fmt.Sprintf("tick %d", i)))
		time.Sleep(30 * time.Millisecond) // long enough for the healthy writers to drain their queues
		n, dropped := hub.Stats()
		fmt.Printf("after tick %d: clients=%d dropped=%d queued-for-stuck=%d\n", i, n, dropped, len(stuck.send))
	}
}

func waitFor(cond func() bool) {
	for i := 0; i < 200 && !cond(); i++ {
		time.Sleep(10 * time.Millisecond)
	}
}

client 1 heard "hello everyone" err=<nil>
client 2 heard "hello everyone" err=<nil>
client 3 heard "hello everyone" err=<nil>
bad origin: status=403 err=websocket: bad handshake
oversize message: websocket: close 1009 (message too big)
after tick 1: clients=3 dropped=0 queued-for-stuck=1
after tick 2: clients=3 dropped=0 queued-for-stuck=2
after tick 3: clients=3 dropped=0 queued-for-stuck=3
after tick 4: clients=3 dropped=0 queued-for-stuck=4
after tick 5: clients=2 dropped=1 queued-for-stuck=4
after tick 6: clients=2 dropped=1 queued-for-stuck=4

Reading the output: all three clients heard the broadcast, including the sender, because the hub sends to everyone in the set; filter by sender if you do not want an echo. The handshake from https://evil.example came back as 403, before any WebSocket existed. The 2,000 byte message made the server close that connection, and the client's next read reported close code 1009, "message too big"; the hub's client count dropped from three to two without us doing anything, since the reader's defer handles cleanup. And the simulated stuck client held four queued messages, then on the fifth was deleted and counted as dropped. The two healthy clients stayed registered throughout, which is the entire point.

I should be honest about the first version of that demo. It broadcast all six ticks in a tight loop with no pause, and the run ended with zero clients and three drops, healthy ones included. Nothing was wrong with the clients. Six messages arrived faster than the two writePump goroutines could be scheduled, so their four slot queues overflowed too. That is the real cost of the policy: a queue size is a statement about the largest burst you will absorb, and "drop on full" cannot tell a slow peer from a fast one that was briefly out of CPU. The final program sleeps 30 ms between ticks, and the numbers above are from that version. In a real service, size the queue for your largest burst (a snapshot of a room's state sent on join is a classic), measure how often drops happen, and treat a rising count as a signal about the queue before you blame the network.

The stuck client in step 4 is a stand-in: a real one has a writePump that blocks inside WriteMessage when the peer stops reading and its TCP buffers fill. On loopback that takes megabytes, which would make a deterministic demonstration slow, so the test fakes the end state directly. The queue behavior it shows is the real one. The write deadline is what eventually frees a real stuck writer: after writeWait the blocked write fails, the pump returns, and the connection closes.

What a connection costs

Each connection holds a file descriptor, two goroutines, a read buffer and a write buffer. With the 1,024 byte buffers configured here, the library's own allocations are small, and the goroutine stacks start at a few kilobytes. In my experience the number that surprises people is not memory but file descriptors: the default soft limit on many Linux systems is 1,024, so a server holding a few thousand sockets hits too many open files long before it runs out of RAM. Raise ulimit -n (or LimitNOFILE in the systemd unit) as part of the deployment, and log the count of open clients as a metric so the number is visible before it is a problem.

Idle connections are cheap in Go for the reason covered in the goroutines article: a parked goroutine costs memory and no CPU. What costs CPU is the fan-out. A broadcast to 10,000 clients is 10,000 channel sends plus 10,000 socket writes, per message. If the same payload goes to everyone, build it once as a []byte (the hub above shares one slice among all queues, which is safe because nobody mutates it) and use websocket.NewPreparedMessage so compression and framing happen once rather than once per recipient.

Rooms, users and message types

The hub above has one room. Real applications need rooms or topics, and the natural extension is a map[string]map[*Client]struct{} guarded by the same mutex, with Join(room, c) and BroadcastTo(room, msg). Keep one rule: a client's own bookkeeping (which rooms it joined) lives on the client, so Remove can leave every room in one pass when the connection dies. Forgetting that is how a hub leaks entries for users who closed their laptop lid.

For message types, I use a small JSON envelope: {"type":"chat","room":"general","body":"..."}. Decode it with a struct and a switch on type, reject unknown types and oversized bodies, and answer errors with an error message to that one client rather than closing the connection, so a typo in the front end does not turn into a reconnect storm. The same discipline applies as for decoding a request body.

What I would add before shipping

Authentication happens on the HTTP request, before the upgrade. Browsers cannot set an Authorization header on a WebSocket, so use a session cookie (with the origin check doing CSRF work, since cookies ride along on cross-site handshakes) or a short-lived ticket obtained from a normal authenticated endpoint and passed as a query parameter. Never log the full URL if it carries a ticket. The auth middleware article covers the session side.

Per-connection rate limiting. A client allowed to send 10,000 messages a second can saturate the hub. Reuse the limiter from rate limiting in middleware per connection inside readPump, and close on abuse.

Graceful shutdown. http.Server.Shutdown does not wait for hijacked connections, because after the upgrade they are not the server's any more. Keep the hub's client set and close every client on shutdown, with a close frame, so browsers reconnect to another instance instead of seeing a reset. The pattern is in graceful shutdown for the HTTP server.

Message framing. Plain text in, plain text out is a demo. Real applications send JSON envelopes with a type field and decode with json.Decoder and DisallowUnknownFields. Validate everything a client sends exactly as you would a request body, because it has the same trust level.

Scaling past one process. The hub lives in memory, so two instances behind a load balancer have two separate rooms. The fix is a pub/sub layer (Redis PUBLISH or NATS) that each instance subscribes to and forwards into its local hub. Do not build that until one machine is genuinely not enough; a single Go process holds tens of thousands of idle WebSockets.

Testing a hub

Test the Hub without any network: create Client values with small channels, broadcast, and assert on the queue contents, as the stuck client above does. Prefer a table of small cases (queue size, number of messages, expected drops) over one big scenario, since each case then explains itself when it fails. Keep one or two end-to-end tests with httptest.NewServer and a real dialer for the handshake, the origin rule and the read limit, and run all of it with -race. If you write a pump test that sleeps, wait on a condition like the waitFor helper above instead, so the test is fast when the code is right and fails loudly when it is not.

If you are choosing between libraries again later, coder/websocket has a context based API and wsjson helpers that remove some of the deadline bookkeeping above; the hub, the bounded queues and the origin policy would not change at all. A last note on choosing the library. The mistakes above, a shared write, a blocking broadcast, a missing deadline, are independent of which package you import. Learn the structure once and either package will do.