conference: fix unsoundness in channel usage

Author: daniel-abramov

pkg/common/channel.go

@@ -6,7 +6,7 @@ import "sync/atomic"
 // Unlike traditional Go channels, these allow the receiver to mark the channel as closed which would then fail
 // to send any messages to the channel over `Send“.
 func NewChannel[M any]() (Sender[M], Receiver[M]) {
-	channel := make(chan M)
+	channel := make(chan M, 128)
 	closed := &atomic.Bool{}
 	sender := Sender[M]{channel, closed}
 	receiver := Receiver[M]{channel, closed}

pkg/conference/processor.go

@@ -29,18 +29,19 @@ func (c *Conference) processMessages() {
 			c.matrixMessages.Close()
 
 			// Let's read remaining messages from the channel (otherwise the caller will be
-			// blocked in case of unbuffered channels).
-			var message *MatrixMessage
-			select {
-			case msg := <-c.matrixMessages.Channel:
-				*message = msg
-			default:
-				// Ok, no messages in the queue, nice.
+			// blocked in case of unbuffered channels). We must read **all** pending messages.
+			messages := make([]MatrixMessage, 0)
+			for {
+				msg, ok := <-c.matrixMessages.Channel
+				if !ok {
+					break
+				}
+				messages = append(messages, msg)
 			}
 
 			// Send the information that we ended to the owner and pass the message
 			// that we did not process (so that we don't drop it silently).
-			c.endNotifier.Notify(message)
+			c.endNotifier.Notify(messages)
 			return
 		}
 	}

pkg/conference/start.go

@@ -43,7 +43,7 @@ func StartConference(
 		matrixMessages: receiver,
 		endNotifier:    conferenceEndNotifier,
 		participants:   make(map[ParticipantID]*Participant),
-		peerMessages:   make(chan common.Message[ParticipantID, peer.MessageContent]),
+		peerMessages:   make(chan common.Message[ParticipantID, peer.MessageContent], 128),
 		logger:         logrus.WithFields(logrus.Fields{"conf_id": confID}),
 	}
 
@@ -60,5 +60,5 @@ func StartConference(
 
 type ConferenceEndNotifier interface {
 	// Called when the conference ends.
-	Notify(unread *MatrixMessage)
+	Notify(unread []MatrixMessage)
 }

pkg/router.go

@@ -44,7 +44,7 @@ func newRouter(matrix *signaling.MatrixClient, config conf.Config) chan<- Router
 		matrix:          matrix,
 		conferenceSinks: make(map[string]*common.Sender[conf.MatrixMessage]),
 		config:          config,
-		channel:         make(chan RouterMessage),
+		channel:         make(chan RouterMessage, 128),
 	}
 
 	// Start the main loop of the Router.
@@ -59,9 +59,9 @@ func newRouter(matrix *signaling.MatrixClient, config conf.Config) chan<- Router
 				// Remove the conference that ended from the list.
 				delete(router.conferenceSinks, msg.conferenceID)
 				// Process the message that was not read by the conference.
-				if msg.unread != nil {
-					// TODO: We must handle this message to avoid glare on session end.
-					// router.handleMatrixEvent(*msg.unread)
+				if len(msg.unread) > 0 {
+					// FIXME: We must handle these messages!
+					logrus.Warnf("Unread messages: %v", len(msg.unread))
 				}
 			}
 		}
@@ -162,7 +162,7 @@ type ConferenceEndedMessage struct {
 	// The ID of the conference that has ended.
 	conferenceID string
 	// A message (or messages in future) that has not been processed (if any).
-	unread *conf.MatrixMessage
+	unread []conf.MatrixMessage
 }
 
 // A simple wrapper around channel that contains the ID of the conference that sent the message.
@@ -180,7 +180,7 @@ func createConferenceEndNotifier(conferenceID string, channel chan<- RouterMessa
 }
 
 // A function that a conference calls when it is ended.
-func (c *ConferenceEndNotifier) Notify(unread *conf.MatrixMessage) {
+func (c *ConferenceEndNotifier) Notify(unread []conf.MatrixMessage) {
 	c.channel <- ConferenceEndedMessage{
 		conferenceID: c.conferenceID,
 		unread:       unread,