Merge federationapi, federationsender, signingkeyserver components (#2055)

* Initial federation sender -> federation API refactoring

* Move base into own package, avoids import cycle

* Fix build errors

* Fix tests

* Add signing key server tables

* Try to fold signing key server into federation API

* Fix dendritejs builds

* Update embedded interfaces

* Fix panic, fix lint error

* Update configs, docker

* Rename some things

* Reuse same keyring on the implementing side

* Fix federation tests, `NewBaseDendrite` can accept freeform options

* Fix build

* Update create_db, configs

* Name tables back

* Don't rename federationsender consumer for now

federationapi/queue/destinationqueue.go

@@ -0,0 +1,448 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package queue
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/matrix-org/dendrite/federationapi/statistics"
+	"github.com/matrix-org/dendrite/federationapi/storage"
+	"github.com/matrix-org/dendrite/federationapi/storage/shared"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/process"
+	"github.com/matrix-org/gomatrix"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+	"go.uber.org/atomic"
+)
+
+const (
+	maxPDUsPerTransaction = 50
+	maxEDUsPerTransaction = 50
+	maxPDUsInMemory       = 128
+	maxEDUsInMemory       = 128
+	queueIdleTimeout      = time.Second * 30
+)
+
+// destinationQueue is a queue of events for a single destination.
+// It is responsible for sending the events to the destination and
+// 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
+	rsAPI              api.RoomserverInternalAPI
+	client             *gomatrixserverlib.FederationClient // federation client
+	origin             gomatrixserverlib.ServerName        // origin of requests
+	destination        gomatrixserverlib.ServerName        // destination of requests
+	running            atomic.Bool                         // is the queue worker running?
+	backingOff         atomic.Bool                         // true if we're backing off
+	overflowed         atomic.Bool                         // the queues exceed maxPDUsInMemory/maxEDUsInMemory, so we should consult the database for more
+	statistics         *statistics.ServerStatistics        // statistics about this remote server
+	transactionIDMutex sync.Mutex                          // protects transactionID
+	transactionID      gomatrixserverlib.TransactionID     // last transaction ID if retrying, or "" if last txn was successful
+	notify             chan struct{}                       // interrupts idle wait pending PDUs/EDUs
+	pendingPDUs        []*queuedPDU                        // PDUs waiting to be sent
+	pendingEDUs        []*queuedEDU                        // EDUs waiting to be sent
+	pendingMutex       sync.RWMutex                        // protects pendingPDUs and pendingEDUs
+	interruptBackoff   chan bool                           // interrupts backoff
+}
+
+// Send event adds the event to the pending queue for the destination.
+// If the queue is empty then it starts a background goroutine to
+// start sending events to that destination.
+func (oq *destinationQueue) sendEvent(event *gomatrixserverlib.HeaderedEvent, receipt *shared.Receipt) {
+	if event == nil {
+		logrus.Errorf("attempt to send nil PDU with destination %q", oq.destination)
+		return
+	}
+	// Create a database entry that associates the given PDU NID with
+	// this destination queue. We'll then be able to retrieve the PDU
+	// later.
+	if err := oq.db.AssociatePDUWithDestination(
+		context.TODO(),
+		"",             // TODO: remove this, as we don't need to persist the transaction ID
+		oq.destination, // the destination server name
+		receipt,        // NIDs from federationapi_queue_json table
+	); err != nil {
+		logrus.WithError(err).Errorf("failed to associate PDU %q with destination %q", event.EventID(), oq.destination)
+		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()
+		if len(oq.pendingPDUs) < maxPDUsInMemory {
+			oq.pendingPDUs = append(oq.pendingPDUs, &queuedPDU{
+				pdu:     event,
+				receipt: receipt,
+			})
+		} else {
+			oq.overflowed.Store(true)
+		}
+		oq.pendingMutex.Unlock()
+		// Wake up the queue if it's asleep.
+		oq.wakeQueueIfNeeded()
+		select {
+		case oq.notify <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// sendEDU adds the EDU event to the pending queue for the destination.
+// If the queue is empty then it starts a background goroutine to
+// start sending events to that destination.
+func (oq *destinationQueue) sendEDU(event *gomatrixserverlib.EDU, receipt *shared.Receipt) {
+	if event == nil {
+		logrus.Errorf("attempt to send nil EDU with destination %q", oq.destination)
+		return
+	}
+	// Create a database entry that associates the given PDU NID with
+	// this destination queue. We'll then be able to retrieve the PDU
+	// later.
+	if err := oq.db.AssociateEDUWithDestination(
+		context.TODO(),
+		oq.destination, // the destination server name
+		receipt,        // NIDs from federationapi_queue_json table
+	); err != nil {
+		logrus.WithError(err).Errorf("failed to associate EDU with destination %q", oq.destination)
+		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()
+		if len(oq.pendingEDUs) < maxEDUsInMemory {
+			oq.pendingEDUs = append(oq.pendingEDUs, &queuedEDU{
+				edu:     event,
+				receipt: receipt,
+			})
+		} else {
+			oq.overflowed.Store(true)
+		}
+		oq.pendingMutex.Unlock()
+		// Wake up the queue if it's asleep.
+		oq.wakeQueueIfNeeded()
+		select {
+		case oq.notify <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// wakeQueueIfNeeded will wake up the destination queue if it is
+// not already running. If it is running but it is backing off
+// then we will interrupt the backoff, causing any federation
+// requests to retry.
+func (oq *destinationQueue) wakeQueueIfNeeded() {
+	// If we are backing off then interrupt the backoff.
+	if oq.backingOff.CAS(true, false) {
+		oq.interruptBackoff <- true
+	}
+	// If we aren't running then wake up the queue.
+	if !oq.running.Load() {
+		// Start the queue.
+		go oq.backgroundSend()
+	}
+}
+
+// getPendingFromDatabase will look at the database and see if
+// there are any persisted events that haven't been sent to this
+// destination yet. If so, they will be queued up.
+func (oq *destinationQueue) getPendingFromDatabase() {
+	// Check to see if there's anything to do for this server
+	// in the database.
+	retrieved := false
+	ctx := context.Background()
+	oq.pendingMutex.Lock()
+	defer oq.pendingMutex.Unlock()
+
+	// Take a note of all of the PDUs and EDUs that we already
+	// have cached. We will index them based on the receipt,
+	// which ultimately just contains the index of the PDU/EDU
+	// in the database.
+	gotPDUs := map[string]struct{}{}
+	gotEDUs := map[string]struct{}{}
+	for _, pdu := range oq.pendingPDUs {
+		gotPDUs[pdu.receipt.String()] = struct{}{}
+	}
+	for _, edu := range oq.pendingEDUs {
+		gotEDUs[edu.receipt.String()] = struct{}{}
+	}
+
+	if pduCapacity := maxPDUsInMemory - len(oq.pendingPDUs); pduCapacity > 0 {
+		// We have room in memory for some PDUs - let's request no more than that.
+		if pdus, err := oq.db.GetPendingPDUs(ctx, oq.destination, pduCapacity); err == nil {
+			for receipt, pdu := range pdus {
+				if _, ok := gotPDUs[receipt.String()]; ok {
+					continue
+				}
+				oq.pendingPDUs = append(oq.pendingPDUs, &queuedPDU{receipt, pdu})
+				retrieved = true
+			}
+		} else {
+			logrus.WithError(err).Errorf("Failed to get pending PDUs for %q", oq.destination)
+		}
+	}
+	if eduCapacity := maxEDUsInMemory - len(oq.pendingEDUs); eduCapacity > 0 {
+		// We have room in memory for some EDUs - let's request no more than that.
+		if edus, err := oq.db.GetPendingEDUs(ctx, oq.destination, eduCapacity); err == nil {
+			for receipt, edu := range edus {
+				if _, ok := gotEDUs[receipt.String()]; ok {
+					continue
+				}
+				oq.pendingEDUs = append(oq.pendingEDUs, &queuedEDU{receipt, edu})
+				retrieved = true
+			}
+		} else {
+			logrus.WithError(err).Errorf("Failed to get pending EDUs for %q", oq.destination)
+		}
+	}
+	// If we've retrieved all of the events from the database with room to spare
+	// in memory then we'll no longer consider this queue to be overflowed.
+	if len(oq.pendingPDUs) < maxPDUsInMemory && len(oq.pendingEDUs) < maxEDUsInMemory {
+		oq.overflowed.Store(false)
+	}
+	// If we've retrieved some events then notify the destination queue goroutine.
+	if retrieved {
+		select {
+		case oq.notify <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// backgroundSend is the worker goroutine for sending events.
+func (oq *destinationQueue) backgroundSend() {
+	// Check if a worker is already running, and if it isn't, then
+	// mark it as started.
+	if !oq.running.CAS(false, true) {
+		return
+	}
+	destinationQueueRunning.Inc()
+	defer destinationQueueRunning.Dec()
+	defer oq.queues.clearQueue(oq)
+	defer oq.running.Store(false)
+
+	// Mark the queue as overflowed, so we will consult the database
+	// to see if there's anything new to send.
+	oq.overflowed.Store(true)
+
+	for {
+		// If we are overflowing memory and have sent things out to the
+		// database then we can look up what those things are.
+		if oq.overflowed.Load() {
+			oq.getPendingFromDatabase()
+		}
+
+		// If we have nothing to do then wait either for incoming events, or
+		// until we hit an idle timeout.
+		select {
+		case <-oq.notify:
+			// There's work to do, either because getPendingFromDatabase
+			// told us there is, or because a new event has come in via
+			// sendEvent/sendEDU.
+		case <-time.After(queueIdleTimeout):
+			// The worker is idle so stop the goroutine. It'll get
+			// restarted automatically the next time we have an event to
+			// send.
+			return
+		}
+
+		// If we are backing off this server then wait for the
+		// backoff duration to complete first, or until explicitly
+		// told to retry.
+		until, blacklisted := oq.statistics.BackoffInfo()
+		if blacklisted {
+			// It's been suggested that we should give up because the backoff
+			// has exceeded a maximum allowable value. Clean up the in-memory
+			// buffers at this point. The PDU clean-up is already on a defer.
+			logrus.Warnf("Blacklisting %q due to exceeding backoff threshold", oq.destination)
+			oq.pendingMutex.Lock()
+			for i := range oq.pendingPDUs {
+				oq.pendingPDUs[i] = nil
+			}
+			for i := range oq.pendingEDUs {
+				oq.pendingEDUs[i] = nil
+			}
+			oq.pendingPDUs = nil
+			oq.pendingEDUs = nil
+			oq.pendingMutex.Unlock()
+			return
+		}
+		if until != nil && until.After(time.Now()) {
+			// We haven't backed off yet, so wait for the suggested amount of
+			// time.
+			duration := time.Until(*until)
+			logrus.Warnf("Backing off %q for %s", oq.destination, duration)
+			oq.backingOff.Store(true)
+			destinationQueueBackingOff.Inc()
+			select {
+			case <-time.After(duration):
+			case <-oq.interruptBackoff:
+			}
+			destinationQueueBackingOff.Dec()
+			oq.backingOff.Store(false)
+		}
+
+		// Work out which PDUs/EDUs to include in the next transaction.
+		oq.pendingMutex.RLock()
+		pduCount := len(oq.pendingPDUs)
+		eduCount := len(oq.pendingEDUs)
+		if pduCount > maxPDUsPerTransaction {
+			pduCount = maxPDUsPerTransaction
+		}
+		if eduCount > maxEDUsPerTransaction {
+			eduCount = maxEDUsPerTransaction
+		}
+		toSendPDUs := oq.pendingPDUs[:pduCount]
+		toSendEDUs := oq.pendingEDUs[:eduCount]
+		oq.pendingMutex.RUnlock()
+
+		// If we have pending PDUs or EDUs then construct a transaction.
+		// Try sending the next transaction and see what happens.
+		transaction, pc, ec, terr := oq.nextTransaction(toSendPDUs, toSendEDUs)
+		if terr != nil {
+			// We failed to send the transaction. Mark it as a failure.
+			oq.statistics.Failure()
+
+		} else if transaction {
+			// If we successfully sent the transaction then clear out
+			// the pending events and EDUs, and wipe our transaction ID.
+			oq.statistics.Success()
+			oq.pendingMutex.Lock()
+			for i := range oq.pendingPDUs[:pc] {
+				oq.pendingPDUs[i] = nil
+			}
+			for i := range oq.pendingEDUs[:ec] {
+				oq.pendingEDUs[i] = nil
+			}
+			oq.pendingPDUs = oq.pendingPDUs[pc:]
+			oq.pendingEDUs = oq.pendingEDUs[ec:]
+			oq.pendingMutex.Unlock()
+		}
+	}
+}
+
+// nextTransaction creates a new transaction from the pending event
+// queue and sends it. Returns true if a transaction was sent or
+// false otherwise.
+func (oq *destinationQueue) nextTransaction(
+	pdus []*queuedPDU,
+	edus []*queuedEDU,
+) (bool, int, int, error) {
+	// If there's no projected transaction ID then generate one. If
+	// the transaction succeeds then we'll set it back to "" so that
+	// we generate a new one next time. If it fails, we'll preserve
+	// it so that we retry with the same transaction ID.
+	oq.transactionIDMutex.Lock()
+	if oq.transactionID == "" {
+		now := gomatrixserverlib.AsTimestamp(time.Now())
+		oq.transactionID = gomatrixserverlib.TransactionID(fmt.Sprintf("%d-%d", now, oq.statistics.SuccessCount()))
+	}
+	oq.transactionIDMutex.Unlock()
+
+	// Create the transaction.
+	t := gomatrixserverlib.Transaction{
+		PDUs: []json.RawMessage{},
+		EDUs: []gomatrixserverlib.EDU{},
+	}
+	t.Origin = oq.origin
+	t.Destination = oq.destination
+	t.OriginServerTS = gomatrixserverlib.AsTimestamp(time.Now())
+	t.TransactionID = oq.transactionID
+
+	// If we didn't get anything from the database and there are no
+	// pending EDUs then there's nothing to do - stop here.
+	if len(pdus) == 0 && len(edus) == 0 {
+		return false, 0, 0, nil
+	}
+
+	var pduReceipts []*shared.Receipt
+	var eduReceipts []*shared.Receipt
+
+	// Go through PDUs that we retrieved from the database, if any,
+	// and add them into the transaction.
+	for _, pdu := range pdus {
+		if pdu == nil || pdu.pdu == nil {
+			continue
+		}
+		// Append the JSON of the event, since this is a json.RawMessage type in the
+		// gomatrixserverlib.Transaction struct
+		t.PDUs = append(t.PDUs, pdu.pdu.JSON())
+		pduReceipts = append(pduReceipts, pdu.receipt)
+	}
+
+	// Do the same for pending EDUS in the queue.
+	for _, edu := range edus {
+		if edu == nil || edu.edu == nil {
+			continue
+		}
+		t.EDUs = append(t.EDUs, *edu.edu)
+		eduReceipts = append(eduReceipts, edu.receipt)
+	}
+
+	logrus.WithField("server_name", oq.destination).Debugf("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs))
+
+	// Try to send the transaction to the destination server.
+	// TODO: we should check for 500-ish fails vs 400-ish here,
+	// since we shouldn't queue things indefinitely in response
+	// to a 400-ish error
+	ctx, cancel := context.WithTimeout(oq.process.Context(), time.Minute*5)
+	defer cancel()
+	_, err := oq.client.SendTransaction(ctx, t)
+	switch err.(type) {
+	case nil:
+		// Clean up the transaction in the database.
+		if pduReceipts != nil {
+			//logrus.Infof("Cleaning PDUs %q", pduReceipt.String())
+			if err = oq.db.CleanPDUs(context.Background(), oq.destination, pduReceipts); err != nil {
+				logrus.WithError(err).Errorf("Failed to clean PDUs for server %q", t.Destination)
+			}
+		}
+		if eduReceipts != nil {
+			//logrus.Infof("Cleaning EDUs %q", eduReceipt.String())
+			if err = oq.db.CleanEDUs(context.Background(), oq.destination, eduReceipts); err != nil {
+				logrus.WithError(err).Errorf("Failed to clean EDUs for server %q", t.Destination)
+			}
+		}
+		// Reset the transaction ID.
+		oq.transactionIDMutex.Lock()
+		oq.transactionID = ""
+		oq.transactionIDMutex.Unlock()
+		return true, len(t.PDUs), len(t.EDUs), nil
+	case gomatrix.HTTPError:
+		// Report that we failed to send the transaction and we
+		// will retry again, subject to backoff.
+		return false, 0, 0, err
+	default:
+		logrus.WithFields(logrus.Fields{
+			"destination":   oq.destination,
+			logrus.ErrorKey: err,
+		}).Debugf("Failed to send transaction %q", t.TransactionID)
+		return false, 0, 0, err
+	}
+}

federationapi/queue/queue.go

@@ -0,0 +1,339 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package queue
+
+import (
+	"context"
+	"crypto/ed25519"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/matrix-org/dendrite/federationapi/statistics"
+	"github.com/matrix-org/dendrite/federationapi/storage"
+	"github.com/matrix-org/dendrite/federationapi/storage/shared"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/process"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
+	"github.com/tidwall/gjson"
+)
+
+// OutgoingQueues is a collection of queues for sending transactions to other
+// matrix servers
+type OutgoingQueues struct {
+	db          storage.Database
+	process     *process.ProcessContext
+	disabled    bool
+	rsAPI       api.RoomserverInternalAPI
+	origin      gomatrixserverlib.ServerName
+	client      *gomatrixserverlib.FederationClient
+	statistics  *statistics.Statistics
+	signing     *SigningInfo
+	queuesMutex sync.Mutex // protects the below
+	queues      map[gomatrixserverlib.ServerName]*destinationQueue
+}
+
+func init() {
+	prometheus.MustRegister(
+		destinationQueueTotal, destinationQueueRunning,
+		destinationQueueBackingOff,
+	)
+}
+
+var destinationQueueTotal = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: "dendrite",
+		Subsystem: "federationapi",
+		Name:      "destination_queues_total",
+	},
+)
+
+var destinationQueueRunning = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: "dendrite",
+		Subsystem: "federationapi",
+		Name:      "destination_queues_running",
+	},
+)
+
+var destinationQueueBackingOff = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: "dendrite",
+		Subsystem: "federationapi",
+		Name:      "destination_queues_backing_off",
+	},
+)
+
+// NewOutgoingQueues makes a new OutgoingQueues
+func NewOutgoingQueues(
+	db storage.Database,
+	process *process.ProcessContext,
+	disabled bool,
+	origin gomatrixserverlib.ServerName,
+	client *gomatrixserverlib.FederationClient,
+	rsAPI api.RoomserverInternalAPI,
+	statistics *statistics.Statistics,
+	signing *SigningInfo,
+) *OutgoingQueues {
+	queues := &OutgoingQueues{
+		disabled:   disabled,
+		process:    process,
+		db:         db,
+		rsAPI:      rsAPI,
+		origin:     origin,
+		client:     client,
+		statistics: statistics,
+		signing:    signing,
+		queues:     map[gomatrixserverlib.ServerName]*destinationQueue{},
+	}
+	// Look up which servers we have pending items for and then rehydrate those queues.
+	if !disabled {
+		time.AfterFunc(time.Second*5, func() {
+			serverNames := map[gomatrixserverlib.ServerName]struct{}{}
+			if names, err := db.GetPendingPDUServerNames(context.Background()); err == nil {
+				for _, serverName := range names {
+					serverNames[serverName] = struct{}{}
+				}
+			} else {
+				log.WithError(err).Error("Failed to get PDU server names for destination queue hydration")
+			}
+			if names, err := db.GetPendingEDUServerNames(context.Background()); err == nil {
+				for _, serverName := range names {
+					serverNames[serverName] = struct{}{}
+				}
+			} else {
+				log.WithError(err).Error("Failed to get EDU server names for destination queue hydration")
+			}
+			for serverName := range serverNames {
+				if queue := queues.getQueue(serverName); queue != nil {
+					queue.wakeQueueIfNeeded()
+				}
+			}
+		})
+	}
+	return queues
+}
+
+// TODO: Move this somewhere useful for other components as we often need to ferry these 3 variables
+// around together
+type SigningInfo struct {
+	ServerName gomatrixserverlib.ServerName
+	KeyID      gomatrixserverlib.KeyID
+	PrivateKey ed25519.PrivateKey
+}
+
+type queuedPDU struct {
+	receipt *shared.Receipt
+	pdu     *gomatrixserverlib.HeaderedEvent
+}
+
+type queuedEDU struct {
+	receipt *shared.Receipt
+	edu     *gomatrixserverlib.EDU
+}
+
+func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *destinationQueue {
+	if oqs.statistics.ForServer(destination).Blacklisted() {
+		return nil
+	}
+	oqs.queuesMutex.Lock()
+	defer oqs.queuesMutex.Unlock()
+	oq, ok := oqs.queues[destination]
+	if !ok || oq != nil {
+		destinationQueueTotal.Inc()
+		oq = &destinationQueue{
+			queues:           oqs,
+			db:               oqs.db,
+			process:          oqs.process,
+			rsAPI:            oqs.rsAPI,
+			origin:           oqs.origin,
+			destination:      destination,
+			client:           oqs.client,
+			statistics:       oqs.statistics.ForServer(destination),
+			notify:           make(chan struct{}, 1),
+			interruptBackoff: make(chan bool),
+			signing:          oqs.signing,
+		}
+		oqs.queues[destination] = oq
+	}
+	return oq
+}
+
+func (oqs *OutgoingQueues) clearQueue(oq *destinationQueue) {
+	oqs.queuesMutex.Lock()
+	defer oqs.queuesMutex.Unlock()
+
+	delete(oqs.queues, oq.destination)
+	destinationQueueTotal.Dec()
+}
+
+type ErrorFederationDisabled struct {
+	Message string
+}
+
+func (e *ErrorFederationDisabled) Error() string {
+	return e.Message
+}
+
+// SendEvent sends an event to the destinations
+func (oqs *OutgoingQueues) SendEvent(
+	ev *gomatrixserverlib.HeaderedEvent, origin gomatrixserverlib.ServerName,
+	destinations []gomatrixserverlib.ServerName,
+) error {
+	if oqs.disabled {
+		return &ErrorFederationDisabled{
+			Message: "Federation disabled",
+		}
+	}
+	if origin != oqs.origin {
+		// TODO: Support virtual hosting; gh issue #577.
+		return fmt.Errorf(
+			"sendevent: unexpected server to send as: got %q expected %q",
+			origin, oqs.origin,
+		)
+	}
+
+	// Deduplicate destinations and remove the origin from the list of
+	// destinations just to be sure.
+	destmap := map[gomatrixserverlib.ServerName]struct{}{}
+	for _, d := range destinations {
+		destmap[d] = struct{}{}
+	}
+	delete(destmap, oqs.origin)
+
+	// Check if any of the destinations are prohibited by server ACLs.
+	for destination := range destmap {
+		if api.IsServerBannedFromRoom(
+			context.TODO(),
+			oqs.rsAPI,
+			ev.RoomID(),
+			destination,
+		) {
+			delete(destmap, destination)
+		}
+	}
+
+	// If there are no remaining destinations then give up.
+	if len(destmap) == 0 {
+		return nil
+	}
+
+	log.WithFields(log.Fields{
+		"destinations": len(destmap), "event": ev.EventID(),
+	}).Infof("Sending event")
+
+	headeredJSON, err := json.Marshal(ev)
+	if err != nil {
+		return fmt.Errorf("json.Marshal: %w", err)
+	}
+
+	nid, err := oqs.db.StoreJSON(context.TODO(), string(headeredJSON))
+	if err != nil {
+		return fmt.Errorf("sendevent: oqs.db.StoreJSON: %w", err)
+	}
+
+	for destination := range destmap {
+		if queue := oqs.getQueue(destination); queue != nil {
+			queue.sendEvent(ev, nid)
+		}
+	}
+
+	return nil
+}
+
+// SendEDU sends an EDU event to the destinations.
+func (oqs *OutgoingQueues) SendEDU(
+	e *gomatrixserverlib.EDU, origin gomatrixserverlib.ServerName,
+	destinations []gomatrixserverlib.ServerName,
+) error {
+	if oqs.disabled {
+		return &ErrorFederationDisabled{
+			Message: "Federation disabled",
+		}
+	}
+	if origin != oqs.origin {
+		// TODO: Support virtual hosting; gh issue #577.
+		return fmt.Errorf(
+			"sendevent: unexpected server to send as: got %q expected %q",
+			origin, oqs.origin,
+		)
+	}
+
+	// Deduplicate destinations and remove the origin from the list of
+	// destinations just to be sure.
+	destmap := map[gomatrixserverlib.ServerName]struct{}{}
+	for _, d := range destinations {
+		destmap[d] = struct{}{}
+	}
+	delete(destmap, oqs.origin)
+
+	// There is absolutely no guarantee that the EDU will have a room_id
+	// field, as it is not required by the spec. However, if it *does*
+	// (e.g. typing notifications) then we should try to make sure we don't
+	// bother sending them to servers that are prohibited by the server
+	// ACLs.
+	if result := gjson.GetBytes(e.Content, "room_id"); result.Exists() {
+		for destination := range destmap {
+			if api.IsServerBannedFromRoom(
+				context.TODO(),
+				oqs.rsAPI,
+				result.Str,
+				destination,
+			) {
+				delete(destmap, destination)
+			}
+		}
+	}
+
+	// If there are no remaining destinations then give up.
+	if len(destmap) == 0 {
+		return nil
+	}
+
+	log.WithFields(log.Fields{
+		"destinations": len(destmap), "edu_type": e.Type,
+	}).Info("Sending EDU event")
+
+	ephemeralJSON, err := json.Marshal(e)
+	if err != nil {
+		return fmt.Errorf("json.Marshal: %w", err)
+	}
+
+	nid, err := oqs.db.StoreJSON(context.TODO(), string(ephemeralJSON))
+	if err != nil {
+		return fmt.Errorf("sendevent: oqs.db.StoreJSON: %w", err)
+	}
+
+	for destination := range destmap {
+		if queue := oqs.getQueue(destination); queue != nil {
+			queue.sendEDU(e, nid)
+		}
+	}
+
+	return nil
+}
+
+// RetryServer attempts to resend events to the given server if we had given up.
+func (oqs *OutgoingQueues) RetryServer(srv gomatrixserverlib.ServerName) {
+	if oqs.disabled {
+		return
+	}
+	if queue := oqs.getQueue(srv); queue != nil {
+		queue.wakeQueueIfNeeded()
+	}
+}