• fix: fix Listen/Unlisten race on Pubsub #15315 👈
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @spikecurtis and the rest of your teammates on Graphite

Yeah, I'm not feeling like it's a great design, but I also couldn't come up with anything better.

2 designs I rejected:

1. I considered having the p.queues be the source of truth with a background goroutine reconciling the Listen/Unlisten, sort of like we do elsewhere in the product, but the problem is that when you Subscribe, you don't want to return until you know the pubsub is listening. So, that kind of eventually consistent model won't work.

2. We could go back to how things were and hold the p.qMu while calling Listen/Unlisten. We'd need to ensure that we keep reading from the notification channel, even if the p.qMu is locked. One way to do that is by creating an unbounded FIFO queue of notifications, so we could keep reading from the channel, even if we can't immediately process the notifications due to the lock. An unbounded queue sounds dangerous though.

Yeah, I think you were right to reject those options, what you ended up is still better.

My idea was to have a single goroutine responsible for coordinating sub/unsub, basically anything queues related. We'd still have to solve for not doing blocking operations synchronously, though.

Could look something like this:

func (p *PGPubsub) listenLoop() {
	var queues map[string]... // p.queues no longer needs locking
	for {
		r := <-p.listen
		switch req.(type) {
		case listen:
			if _, ok := queues[r.event]; !ok {
				// ~go p.pqListener.Listen(); r.err <- err
			} else {
				r.err <- nil
			}
		case unlisten:
			// like listen, but unlisten
		case receive:
			// enqueue notif
	}
}

func (p *PGPubsub) subscribeQueue(event string, newQ *msgQueue) (cancel func(), err error) {
	errc := make(chan error, 1)
	p.listen <- listen{"event", newQ, errc}
	_ = <-errc
	return func() {
		p.listen <- unlisten{"event", newQ, errc}
	}, nil
}