Associate events in db before queueing them to send
Fixes a race condition between sending federation events and having them
fully associated in the database.
MODIFY
federationapi/queue/destinationqueue.go
federationapi/queue/destinationqueue.go
@@ -76,6 +76,9 @@ func (oq *destinationQueue) sendEvent(event *gomatrixserverlib.HeaderedEvent, re
return
}
+ // Check if the destination is blacklisted. If it isn't then wake
+ // up the queue.
+ if !oq.statistics.Blacklisted() {
// If there's room in memory to hold the event then add it to the
// list.
oq.pendingMutex.Lock()
@@ -93,6 +96,7 @@ func (oq *destinationQueue) sendEvent(event *gomatrixserverlib.HeaderedEvent, re
oq.wakeQueueAndNotify()
}
}
+}
// sendEDU adds the EDU event to the pending queue for the destination.
// If the queue is empty then it starts a background goroutine to
@@ -103,6 +107,9 @@ func (oq *destinationQueue) sendEDU(event *gomatrixserverlib.EDU, receipt *share
return
}
+ // Check if the destination is blacklisted. If it isn't then wake
+ // up the queue.
+ if !oq.statistics.Blacklisted() {
// If there's room in memory to hold the event then add it to the
// list.
oq.pendingMutex.Lock()
@@ -120,6 +127,7 @@ func (oq *destinationQueue) sendEDU(event *gomatrixserverlib.EDU, receipt *share
oq.wakeQueueAndNotify()
}
}
+}
// handleBackoffNotifier is registered as the backoff notification
// callback with Statistics. It will wakeup and notify the queue

MODIFY
federationapi/queue/queue.go
federationapi/queue/queue.go
@@ -247,9 +247,10 @@ func (oqs *OutgoingQueues) SendEvent(
return fmt.Errorf("sendevent: oqs.db.StoreJSON: %w", err)
}
+ destQueues := make([]*destinationQueue, 0, len(destmap))
for destination := range destmap {
- if queue := oqs.getQueue(destination); queue != nil && !queue.statistics.Blacklisted() {
- queue.sendEvent(ev, nid)
+ if queue := oqs.getQueue(destination); queue != nil {
+ destQueues = append(destQueues, queue)
} else {
delete(destmap, destination)
}
@@ -267,6 +268,14 @@ func (oqs *OutgoingQueues) SendEvent(
return err
}
+ // NOTE : PDUs should be associated with destinations before sending
+ // them, otherwise this is technically a race.
+ // If the send completes before they are associated then they won't
+ // get properly cleaned up in the database.
+ for _, queue := range destQueues {
+ queue.sendEvent(ev, nid)
+ }
+
return nil
}
@@ -335,20 +344,21 @@ func (oqs *OutgoingQueues) SendEDU(
return fmt.Errorf("sendevent: oqs.db.StoreJSON: %w", err)
}
+ destQueues := make([]*destinationQueue, 0, len(destmap))
for destination := range destmap {
- if queue := oqs.getQueue(destination); queue != nil && !queue.statistics.Blacklisted() {
- queue.sendEDU(e, nid)
+ if queue := oqs.getQueue(destination); queue != nil {
+ destQueues = append(destQueues, queue)
} else {
delete(destmap, destination)
}
}
// Create a database entry that associates the given PDU NID with
- // this destination queue. We'll then be able to retrieve the PDU
+ // these destination queues. We'll then be able to retrieve the PDU
// later.
if err := oqs.db.AssociateEDUWithDestinations(
oqs.process.Context(),
- destmap, // the destination server name
+ destmap, // the destination server names
nid, // NIDs from federationapi_queue_json table
e.Type,
nil, // this will use the default expireEDUTypes map
@@ -357,6 +367,14 @@ func (oqs *OutgoingQueues) SendEDU(
return err
}
+ // NOTE : EDUs should be associated with destinations before sending
+ // them, otherwise this is technically a race.
+ // If the send completes before they are associated then they won't
+ // get properly cleaned up in the database.
+ for _, queue := range destQueues {
+ queue.sendEDU(e, nid)
+ }
+
return nil
}

Associate events in db before queueing them to send (#2833)

Fixes a race condition between sending federation events and having them
fully associated in the database.