• 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 onGraphite

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 thep.queues be the source of truth with a background goroutine reconciling theListen/Unlisten, sort of like we do elsewhere in the product, but the problem is that when youSubscribe, 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 thep.qMu while callingListen/Unlisten. We'd need to ensure that we keep reading from the notification channel, even if thep.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() {varqueuesmap[string]...// p.queues no longer needs lockingfor {r:=<-p.listenswitchreq.(type) {caselisten:if_,ok:=queues[r.event];!ok {// ~go p.pqListener.Listen(); r.err <- err}else {r.err<-nil}caseunlisten:// like listen, but unlistencasereceive:// enqueue notif}}func (p*PGPubsub)subscribeQueue(eventstring,newQ*msgQueue) (cancelfunc(),errerror) {errc:=make(chanerror,1)p.listen<-listen{"event",newQ,errc}_=<-errcreturnfunc() {p.listen<-unlisten{"event",newQ,errc}},nil}