Introduce eventUpstreamDisconnected

This allows us to perform cleanup actions in the user goroutine. This
removes the need for pendingLISTsLock.

downstream.go

@@ -1080,9 +1080,6 @@ func (dc *downstreamConn) handleMessageRegistered(msg *irc.Message) error {
 	case "LIST":
 		// TODO: support ELIST when supported by all upstreams
 
-		dc.user.pendingLISTsLock.Lock()
-		defer dc.user.pendingLISTsLock.Unlock()
-
 		pl := pendingLIST{
 			downstreamID:    dc.id,
 			pendingCommands: make(map[int64]*irc.Message),

upstream.go

@@ -63,7 +63,6 @@ type upstreamConn struct {
 	saslStarted bool
 
 	// set of LIST commands in progress, per downstream
-	// access is synchronized with user.pendingLISTsLock
 	pendingLISTDownstreamSet map[uint64]struct{}
 
 	logs map[string]entityLog
@@ -151,10 +150,6 @@ func (uc *upstreamConn) Close() error {
 		return fmt.Errorf("upstream connection already closed")
 	}
 	close(uc.closed)
-	for _, log := range uc.logs {
-		log.file.Close()
-	}
-	uc.endPendingLists(true)
 	return nil
 }
 
@@ -192,8 +187,6 @@ func (uc *upstreamConn) isChannel(entity string) bool {
 }
 
 func (uc *upstreamConn) getPendingList() *pendingLIST {
-	uc.user.pendingLISTsLock.Lock()
-	defer uc.user.pendingLISTsLock.Unlock()
 	for _, pl := range uc.user.pendingLISTs {
 		if _, ok := pl.pendingCommands[uc.network.ID]; !ok {
 			continue
@@ -205,8 +198,6 @@ func (uc *upstreamConn) getPendingList() *pendingLIST {
 
 func (uc *upstreamConn) endPendingLists(all bool) (found bool) {
 	found = false
-	uc.user.pendingLISTsLock.Lock()
-	defer uc.user.pendingLISTsLock.Unlock()
 	for i := 0; i < len(uc.user.pendingLISTs); i++ {
 		pl := uc.user.pendingLISTs[i]
 		if _, ok := pl.pendingCommands[uc.network.ID]; !ok {

user.go

@@ -14,6 +14,10 @@ type eventUpstreamMessage struct {
 	uc  *upstreamConn
 }
 
+type eventUpstreamDisconnected struct {
+	uc *upstreamConn
+}
+
 type eventDownstreamMessage struct {
 	msg *irc.Message
 	dc  *downstreamConn
@@ -75,6 +79,7 @@ func (net *network) run() {
 			uc.logger.Printf("failed to handle messages: %v", err)
 		}
 		uc.Close()
+		net.user.events <- eventUpstreamDisconnected{uc}
 
 		net.lock.Lock()
 		net.conn = nil
@@ -98,7 +103,6 @@ type user struct {
 	downstreamConns []*downstreamConn
 
 	// LIST commands in progress
-	pendingLISTsLock sync.Mutex
 	pendingLISTs []pendingLIST
 }
 
@@ -163,6 +167,12 @@ func (u *user) run() {
 
 	for e := range u.events {
 		switch e := e.(type) {
+		case eventUpstreamDisconnected:
+			uc := e.uc
+			for _, log := range uc.logs {
+				log.file.Close()
+			}
+			uc.endPendingLists(true)
 		case eventUpstreamMessage:
 			msg, uc := e.msg, e.uc
 			if uc.isClosed() {