go / wakeups
A page that shows the state of a system is stale the moment the browser draws it. A script that pushed a commit and wants the test result has to guess how long to sleep. A worker that wants a job has to ask again and again.
I solve all three with one signal. Postgres says "something changed". Go fans that signal out to whoever is waiting. Each waiter re-reads what it cares about. Nothing polls on a timer, and a reader learns of a write within a second.
The signal carries no data. That is what keeps it cheap. This article covers the server side. web / live regions covers what a page does when the signal arrives.
One signal
Every state a reader cares about is a row in Postgres: a job, a
result, a comment, a deploy. A trigger on each table fires NOTIFY
after any write:
CREATE FUNCTION notify_wakeup ()
RETURNS TRIGGER
AS $$
BEGIN
NOTIFY wakeup;
RETURN NULL;
END;
$$
LANGUAGE plpgsql;
CREATE TRIGGER comment_wakeup
AFTER INSERT OR UPDATE OR DELETE ON comment
EXECUTE PROCEDURE notify_wakeup ();The trigger is statement-level, so a bulk write fires it once.
The notification has no payload. A payload is a second schema, and it drifts from the tables it describes. A reader who gets the bare signal reads the rows it needs, and the rows are the truth. A spurious wakeup costs one small query, so I do not try to suppress them.
The trigger fires from Postgres rather than from the Go code that wrote
the row. Every write path then fires it: the handler, a psql
session, a migration, and code I write next year and forget about.
Postgres also delivers the notification at commit, so a reader who
wakes sees the row.
One listener
A pgxpool connection cannot receive notifications, so the server
opens one plain connection for LISTEN:
func (s *server) listen(ctx context.Context) error {
conn, err := pgx.Connect(ctx, s.dbURL)
if err != nil {
return fmt.Errorf("listener connect: %w", err)
}
defer conn.Close(context.Background())
if _, err := conn.Exec(ctx, "LISTEN wakeup"); err != nil {
return fmt.Errorf("listen: %w", err)
}
if err := s.loadState(ctx); err != nil {
return fmt.Errorf("load state: %w", err)
}
for {
if _, err := conn.WaitForNotification(ctx); err != nil {
return fmt.Errorf("wait notification: %w", err)
}
s.wake()
}
}A loop around it reconnects after a failure, with a pause so a database that is down does not spin the loop. It loads the full state on every connect, before it waits. A notification sent during the disconnect is lost, and the reload covers it.
Fan-out
Inside the process, a subscriber is a channel with a buffer of one:
func (s *server) subscribe() chan struct{} {
ch := make(chan struct{}, 1)
s.subsMu.Lock()
s.subs[ch] = struct{}{}
s.subsMu.Unlock()
return ch
}
func (s *server) unsubscribe(ch chan struct{}) {
s.subsMu.Lock()
delete(s.subs, ch)
s.subsMu.Unlock()
}
func (s *server) wake() {
s.subsMu.Lock()
defer s.subsMu.Unlock()
for ch := range s.subs {
select {
case ch <- struct{}{}:
default:
}
}
}The buffer of one and the non-blocking send do two jobs. A burst of writes coalesces into one pending wakeup, so a sweep of result rows wakes a subscriber once. A subscriber that is slow to read cannot block the listener, so one stuck request cannot stall every other waiter.
The subscriber gets no data from the channel. It re-reads.
Subscribe before the first read:
wake := s.subscribe()
defer s.unsubscribe(wake)
if settled := s.read(); settled {
return
}
<-wakeIn the other order there is a gap between the read and the subscribe. A write that lands in the gap wakes nobody, and the waiter sleeps until the write after it.
A held request
A worker that wants a job sends the state it last saw. The server answers when the state differs, or when a timeout ends. The answer is the current state either way:
func (s *server) longPoll(ctx context.Context, old State) State {
ctx, cancel := context.WithTimeout(ctx, PollWait)
defer cancel()
c := make(chan State, 1)
go func() { c <- s.waitNewState(old) }()
select {
case cur := <-c:
return cur
case <-ctx.Done():
return old
}
}The timeout answers with a state, not with an error. A request held open forever cannot be told from a server that stopped. A request that answers every 25 seconds with "no news" proves the server is up.
PollWait is one constant that both sides import. The client's HTTP
timeout is PollWait plus a margin. A client that gives up first
aborts every idle poll, throws the connection away, and logs a timeout
for a server that behaved.
Replacing sleep
A script that pushes a commit and then asks how the checks went has to guess how long to wait. A guess too long wastes time nobody needed. A guess too short reads a green answer about the commit before the one it pushed. No script gets the number right, because the number is the length of somebody's test suite.
The CLI asks the server to hold the request instead. The server subscribes, checks whether the head has settled, and waits for a wakeup or the deadline before it reads again:
func (s *server) waitForHead(ctx context.Context, c *change, sha string) (*change, error) {
wake := s.subscribe()
defer s.unsubscribe(wake)
deadline := time.NewTimer(PollWait)
defer deadline.Stop()
for {
settled, err := s.headSettled(ctx, c, sha)
if err != nil || settled {
return c, err
}
select {
case <-wake:
case <-deadline.C:
return c, nil
case <-ctx.Done():
return nil, ctx.Err()
}
if c, err = s.findChange(ctx, c.ID); err != nil {
return nil, err
}
}
}The CLI loops on that request until the change settles or its own timeout runs out. An answer that is not the end still carries the checks that finished so far, so the CLI prints one line per check as it lands.
The sha is the commit the CLI stands on. A wait that reaches the
server before the push hook records the new head would read the old
head's checks: green, about the wrong code. The server treats a head
that is not sha yet as not settled. A sleep hid that mistake and
did not fix it.
The browser
The same subscription sits behind a Server-Sent Events endpoint. Each
wakeup writes an empty event named wake, and a comment line every
15 seconds keeps a proxy from closing an idle connection:
func (s *server) wakeStream(w http.ResponseWriter, req *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache, no-store, no-transform")
w.Header().Set("X-Accel-Buffering", "no")
flush := w.(http.Flusher).Flush
flush()
ch := s.subscribe()
defer s.unsubscribe(ch)
keepalive := time.NewTicker(15 * time.Second)
defer keepalive.Stop()
for {
select {
case <-ch:
io.WriteString(w, "event: wake\ndata:\n\n")
case <-keepalive.C:
io.WriteString(w, ": keepalive\n\n")
case <-req.Context().Done():
return
}
flush()
}
}The event says nothing about what changed. The page decides what to re-read, which is the subject of web / live regions.
Fit
This fits a system whose state is small enough to re-read on every
wakeup, and whose writes all go through one Postgres database. The
subscribers live in process memory, and Postgres delivers each
notification to every listening connection, so a second server process
opens its own LISTEN and wakes its own subscribers. There is no
message broker and no schema for events. The cost is a query per
wakeup per waiter, which a small state makes cheap.
I use it in cibot, where the waiters are worker boxes,
cibot show --wait, and every open dashboard page.