Don't hold destination queues in memory forever

federationsender/queue/destinationqueue.go

@@ -46,6 +46,7 @@ const (
 // ensures that only one request is in flight to a given destination
 // at a time.
 type destinationQueue struct {
+	queues             *OutgoingQueues
 	db                 storage.Database
 	process            *process.ProcessContext
 	signing            *SigningInfo
@@ -270,6 +271,7 @@ func (oq *destinationQueue) backgroundSend() {
 			// The worker is idle so stop the goroutine. It'll get
 			// restarted automatically the next time we have an event to
 			// send.
+			oq.queues.clearQueue(oq.destination)
 			return
 		}
 

federationsender/queue/queue.go

@@ -154,6 +154,7 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
 	if oq == nil {
 		destinationQueueTotal.Inc()
 		oq = &destinationQueue{
+			queues:           oqs,
 			db:               oqs.db,
 			process:          oqs.process,
 			rsAPI:            oqs.rsAPI,
@@ -170,6 +171,22 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
 	return oq
 }
 
+func (oqs *OutgoingQueues) clearQueue(destination gomatrixserverlib.ServerName) {
+	oqs.queuesMutex.Lock()
+	defer oqs.queuesMutex.Unlock()
+	oq, ok := oqs.queues[destination]
+	switch {
+	case !ok:
+		return
+	case oq.running.Load():
+		return
+	case oq.backingOff.Load():
+		return
+	}
+	delete(oqs.queues, destination)
+	destinationQueueTotal.Dec()
+}
+
 type ErrorFederationDisabled struct {
 	Message string
 }