diff --git a/federationsender/queue/destinationqueue.go b/federationsender/queue/destinationqueue.go index 57c8cff6d..45111bf87 100644 --- a/federationsender/queue/destinationqueue.go +++ b/federationsender/queue/destinationqueue.go @@ -22,6 +22,7 @@ import ( "time" "github.com/matrix-org/dendrite/federationsender/storage" + "github.com/matrix-org/dendrite/federationsender/storage/shared" "github.com/matrix-org/dendrite/federationsender/types" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/gomatrix" @@ -31,8 +32,11 @@ import ( "go.uber.org/atomic" ) -const maxPDUsPerTransaction = 50 -const queueIdleTimeout = time.Second * 30 +const ( + maxPDUsPerTransaction = 50 + maxEDUsPerTransaction = 50 + 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 @@ -49,20 +53,20 @@ type destinationQueue struct { backingOff atomic.Bool // true if we're backing off statistics *types.ServerStatistics // statistics about this remote server incomingInvites chan *gomatrixserverlib.InviteV2Request // invites to send - incomingEDUs chan *gomatrixserverlib.EDU // EDUs to send transactionIDMutex sync.Mutex // protects transactionID transactionID gomatrixserverlib.TransactionID // last transaction ID transactionCount atomic.Int32 // how many events in this transaction so far pendingEDUs []*gomatrixserverlib.EDU // owned by backgroundSend pendingInvites []*gomatrixserverlib.InviteV2Request // owned by backgroundSend notifyPDUs chan bool // interrupts idle wait for PDUs + notifyEDUs chan bool // interrupts idle wait for EDUs 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(nid int64) { +func (oq *destinationQueue) sendEvent(receipt *shared.Receipt) { if oq.statistics.Blacklisted() { // If the destination is blacklisted then drop the event. log.Infof("%s is blacklisted; dropping event", oq.destination) @@ -86,9 +90,9 @@ func (oq *destinationQueue) sendEvent(nid int64) { context.TODO(), oq.transactionID, // the current transaction ID oq.destination, // the destination server name - []int64{nid}, // NID from federationsender_queue_json table + receipt, // NIDs from federationsender_queue_json table ); err != nil { - log.WithError(err).Errorf("failed to associate PDU NID %d with destination %q", nid, oq.destination) + log.WithError(err).Errorf("failed to associate PDU receipt %q with destination %q", receipt.String(), oq.destination) return } // We've successfully added a PDU to the transaction so increase @@ -107,13 +111,34 @@ func (oq *destinationQueue) sendEvent(nid int64) { // 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(ev *gomatrixserverlib.EDU) { +func (oq *destinationQueue) sendEDU(receipt *shared.Receipt) { if oq.statistics.Blacklisted() { // If the destination is blacklisted then drop the event. + log.Infof("%s is blacklisted; dropping ephemeral event", 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 federationsender_queue_json table + ); err != nil { + log.WithError(err).Errorf("failed to associate EDU receipt %q with destination %q", receipt.String(), oq.destination) + return + } + // We've successfully added an EDU to the transaction so increase + // the counter. + oq.transactionCount.Add(1) + // Wake up the queue if it's asleep. oq.wakeQueueIfNeeded() - oq.incomingEDUs <- ev + // If we're blocking on waiting PDUs then tell the queue that we + // have work to do. + select { + case oq.notifyEDUs <- true: + default: + } } // sendInvite adds the invite event to the pending queue for the @@ -166,6 +191,28 @@ func (oq *destinationQueue) waitForPDUs() chan bool { return oq.notifyPDUs } +// waitForEDUs returns a channel for pending EDUs, which will be +// used in backgroundSend select. It returns a closed channel if +// there is something pending right now, or an open channel if +// we're waiting for something. +func (oq *destinationQueue) waitForEDUs() chan bool { + pendingEDUs, err := oq.db.GetPendingEDUCount(context.TODO(), oq.destination) + if err != nil { + log.WithError(err).Errorf("Failed to get pending EDU count on queue %q", oq.destination) + } + // If there are EDUs pending right now then we'll return a closed + // channel. This will mean that the backgroundSend will not block. + if pendingEDUs > 0 { + ch := make(chan bool, 1) + close(ch) + return ch + } + // If there are no EDUs pending right now then instead we'll return + // the notify channel, so that backgroundSend can pick up normal + // notifications from sendEvent. + return oq.notifyEDUs +} + // backgroundSend is the worker goroutine for sending events. // nolint:gocyclo func (oq *destinationQueue) backgroundSend() { @@ -177,7 +224,7 @@ func (oq *destinationQueue) backgroundSend() { defer oq.running.Store(false) for { - pendingPDUs := false + pendingPDUs, pendingEDUs := false, false // If we have nothing to do then wait either for incoming events, or // until we hit an idle timeout. @@ -186,18 +233,10 @@ func (oq *destinationQueue) backgroundSend() { // We were woken up because there are new PDUs waiting in the // database. pendingPDUs = true - case edu := <-oq.incomingEDUs: - // EDUs are handled in-memory for now. We will try to keep - // the ordering intact. - // TODO: Certain EDU types need persistence, e.g. send-to-device - oq.pendingEDUs = append(oq.pendingEDUs, edu) - // If there are any more things waiting in the channel queue - // then read them. This is safe because we guarantee only - // having one goroutine per destination queue, so the channel - // isn't being consumed anywhere else. - for len(oq.incomingEDUs) > 0 { - oq.pendingEDUs = append(oq.pendingEDUs, <-oq.incomingEDUs) - } + case <-oq.waitForEDUs(): + // We were woken up because there are new PDUs waiting in the + // database. + pendingEDUs = true case invite := <-oq.incomingInvites: // There's no strict ordering requirement for invites like // there is for transactions, so we put the invite onto the @@ -238,16 +277,15 @@ func (oq *destinationQueue) backgroundSend() { } // If we have pending PDUs or EDUs then construct a transaction. - if pendingPDUs || len(oq.pendingEDUs) > 0 { + if pendingPDUs || pendingEDUs { // Try sending the next transaction and see what happens. - transaction, terr := oq.nextTransaction(oq.pendingEDUs) + transaction, terr := oq.nextTransaction() if terr != nil { // We failed to send the transaction. if giveUp := oq.statistics.Failure(); giveUp { // 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. - oq.cleanPendingEDUs() oq.cleanPendingInvites() log.Infof("Blacklisting %q due to errors", oq.destination) return @@ -265,8 +303,6 @@ func (oq *destinationQueue) backgroundSend() { // If we successfully sent the transaction then clear out // the pending events and EDUs, and wipe our transaction ID. oq.statistics.Success() - // Clean up the in-memory buffers. - oq.cleanPendingEDUs() } } @@ -294,15 +330,6 @@ func (oq *destinationQueue) backgroundSend() { } } -// cleanPendingEDUs cleans out the pending EDU buffer, removing -// all references so that the underlying objects can be GC'd. -func (oq *destinationQueue) cleanPendingEDUs() { - for i := 0; i < len(oq.pendingEDUs); i++ { - oq.pendingEDUs[i] = nil - } - oq.pendingEDUs = []*gomatrixserverlib.EDU{} -} - // cleanPendingInvites cleans out the pending invite buffer, // removing all references so that the underlying objects can // be GC'd. @@ -316,9 +343,7 @@ func (oq *destinationQueue) cleanPendingInvites() { // 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( - pendingEDUs []*gomatrixserverlib.EDU, -) (bool, error) { +func (oq *destinationQueue) nextTransaction() (bool, error) { // Before we do anything, we need to roll over the transaction // ID that is being used to coalesce events into the next TX. // Otherwise it's possible that we'll pick up an incomplete @@ -343,7 +368,7 @@ func (oq *destinationQueue) nextTransaction( // actually retrieve that many events. ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) defer cancel() - txid, pdus, err := oq.db.GetNextTransactionPDUs( + txid, pdus, pduReceipt, err := oq.db.GetNextTransactionPDUs( ctx, // context oq.destination, // server name maxPDUsPerTransaction, // max events to retrieve @@ -353,9 +378,19 @@ func (oq *destinationQueue) nextTransaction( return false, fmt.Errorf("oq.db.GetNextTransactionPDUs: %w", err) } + edus, eduReceipt, err := oq.db.GetNextTransactionEDUs( + ctx, // context + oq.destination, // server name + maxEDUsPerTransaction, // max events to retrieve + ) + if err != nil { + log.WithError(err).Errorf("failed to get next transaction EDUs for server %q", oq.destination) + return false, fmt.Errorf("oq.db.GetNextTransactionEDUs: %w", err) + } + // 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(pendingEDUs) == 0 { + if len(pdus) == 0 && len(edus) == 0 { return false, nil } @@ -377,7 +412,7 @@ func (oq *destinationQueue) nextTransaction( } // Do the same for pending EDUS in the queue. - for _, edu := range pendingEDUs { + for _, edu := range edus { t.EDUs = append(t.EDUs, *edu) } @@ -393,12 +428,17 @@ func (oq *destinationQueue) nextTransaction( switch err.(type) { case nil: // Clean up the transaction in the database. - if err = oq.db.CleanTransactionPDUs( - context.Background(), - t.Destination, - t.TransactionID, - ); err != nil { - log.WithError(err).Errorf("failed to clean transaction %q for server %q", t.TransactionID, t.Destination) + if pduReceipt != nil { + //logrus.Infof("Cleaning PDUs %q", pduReceipt.String()) + if err = oq.db.CleanPDUs(context.Background(), oq.destination, pduReceipt); err != nil { + log.WithError(err).Errorf("failed to clean PDUs %q for server %q", pduReceipt.String(), t.Destination) + } + } + if eduReceipt != nil { + //logrus.Infof("Cleaning EDUs %q", eduReceipt.String()) + if err = oq.db.CleanEDUs(context.Background(), oq.destination, eduReceipt); err != nil { + log.WithError(err).Errorf("failed to clean EDUs %q for server %q", eduReceipt.String(), t.Destination) + } } return true, nil case gomatrix.HTTPError: diff --git a/federationsender/queue/queue.go b/federationsender/queue/queue.go index 812267e63..e488a34aa 100644 --- a/federationsender/queue/queue.go +++ b/federationsender/queue/queue.go @@ -61,12 +61,23 @@ func NewOutgoingQueues( queues: map[gomatrixserverlib.ServerName]*destinationQueue{}, } // Look up which servers we have pending items for and then rehydrate those queues. - if serverNames, err := db.GetPendingPDUServerNames(context.Background()); err == nil { - for _, serverName := range serverNames { - queues.getQueue(serverName).wakeQueueIfNeeded() + 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 server names for destination queue hydration") + 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 { + queues.getQueue(serverName).wakeQueueIfNeeded() } return queues } @@ -91,9 +102,9 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d destination: destination, client: oqs.client, statistics: oqs.statistics.ForServer(destination), - incomingEDUs: make(chan *gomatrixserverlib.EDU, 128), incomingInvites: make(chan *gomatrixserverlib.InviteV2Request, 128), notifyPDUs: make(chan bool, 1), + notifyEDUs: make(chan bool, 1), interruptBackoff: make(chan bool), signing: oqs.signing, } @@ -196,8 +207,18 @@ func (oqs *OutgoingQueues) SendEDU( }).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 destinations { - oqs.getQueue(destination).sendEDU(e) + oqs.getQueue(destination).sendEDU(nid) } return nil diff --git a/federationsender/storage/interface.go b/federationsender/storage/interface.go index a24158033..1bea83e20 100644 --- a/federationsender/storage/interface.go +++ b/federationsender/storage/interface.go @@ -17,6 +17,7 @@ package storage import ( "context" + "github.com/matrix-org/dendrite/federationsender/storage/shared" "github.com/matrix-org/dendrite/federationsender/types" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/gomatrixserverlib" @@ -24,13 +25,26 @@ import ( type Database interface { internal.PartitionStorer + UpdateRoom(ctx context.Context, roomID, oldEventID, newEventID string, addHosts []types.JoinedHost, removeHosts []string) (joinedHosts []types.JoinedHost, err error) + GetJoinedHosts(ctx context.Context, roomID string) ([]types.JoinedHost, error) GetAllJoinedHosts(ctx context.Context) ([]gomatrixserverlib.ServerName, error) - StoreJSON(ctx context.Context, js string) (int64, error) - AssociatePDUWithDestination(ctx context.Context, transactionID gomatrixserverlib.TransactionID, serverName gomatrixserverlib.ServerName, nids []int64) error - GetNextTransactionPDUs(ctx context.Context, serverName gomatrixserverlib.ServerName, limit int) (gomatrixserverlib.TransactionID, []*gomatrixserverlib.HeaderedEvent, error) - CleanTransactionPDUs(ctx context.Context, serverName gomatrixserverlib.ServerName, transactionID gomatrixserverlib.TransactionID) error + + StoreJSON(ctx context.Context, js string) (*shared.Receipt, error) + + AssociatePDUWithDestination(ctx context.Context, transactionID gomatrixserverlib.TransactionID, serverName gomatrixserverlib.ServerName, receipt *shared.Receipt) error + AssociateEDUWithDestination(ctx context.Context, serverName gomatrixserverlib.ServerName, receipt *shared.Receipt) error + + GetNextTransactionPDUs(ctx context.Context, serverName gomatrixserverlib.ServerName, limit int) (gomatrixserverlib.TransactionID, []*gomatrixserverlib.HeaderedEvent, *shared.Receipt, error) + GetNextTransactionEDUs(ctx context.Context, serverName gomatrixserverlib.ServerName, limit int) ([]*gomatrixserverlib.EDU, *shared.Receipt, error) + + CleanPDUs(ctx context.Context, serverName gomatrixserverlib.ServerName, receipt *shared.Receipt) error + CleanEDUs(ctx context.Context, serverName gomatrixserverlib.ServerName, receipt *shared.Receipt) error + GetPendingPDUCount(ctx context.Context, serverName gomatrixserverlib.ServerName) (int64, error) + GetPendingEDUCount(ctx context.Context, serverName gomatrixserverlib.ServerName) (int64, error) + GetPendingPDUServerNames(ctx context.Context) ([]gomatrixserverlib.ServerName, error) + GetPendingEDUServerNames(ctx context.Context) ([]gomatrixserverlib.ServerName, error) } diff --git a/federationsender/storage/postgres/queue_edus_table.go b/federationsender/storage/postgres/queue_edus_table.go index b531907e5..995b04521 100644 --- a/federationsender/storage/postgres/queue_edus_table.go +++ b/federationsender/storage/postgres/queue_edus_table.go @@ -19,6 +19,7 @@ import ( "context" "database/sql" + "github.com/lib/pq" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/gomatrixserverlib" @@ -42,9 +43,13 @@ const insertQueueEDUSQL = "" + "INSERT INTO federationsender_queue_edus (edu_type, server_name, json_nid)" + " VALUES ($1, $2, $3)" +const deleteQueueEDUSQL = "" + + "DELETE FROM federationsender_queue_edus WHERE server_name = $1 AND json_nid = ANY($2)" + const selectQueueEDUSQL = "" + "SELECT json_nid FROM federationsender_queue_edus" + - " WHERE server_name = $1" + " WHERE server_name = $1" + + " LIMIT $2" const selectQueueEDUReferenceJSONCountSQL = "" + "SELECT COUNT(*) FROM federationsender_queue_edus" + @@ -60,6 +65,7 @@ const selectQueueServerNamesSQL = "" + type queueEDUsStatements struct { db *sql.DB insertQueueEDUStmt *sql.Stmt + deleteQueueEDUStmt *sql.Stmt selectQueueEDUStmt *sql.Stmt selectQueueEDUReferenceJSONCountStmt *sql.Stmt selectQueueEDUCountStmt *sql.Stmt @@ -77,6 +83,9 @@ func NewPostgresQueueEDUsTable(db *sql.DB) (s *queueEDUsStatements, err error) { if s.insertQueueEDUStmt, err = s.db.Prepare(insertQueueEDUSQL); err != nil { return } + if s.deleteQueueEDUStmt, err = s.db.Prepare(deleteQueueEDUSQL); err != nil { + return + } if s.selectQueueEDUStmt, err = s.db.Prepare(selectQueueEDUSQL); err != nil { return } @@ -95,26 +104,37 @@ func NewPostgresQueueEDUsTable(db *sql.DB) (s *queueEDUsStatements, err error) { func (s *queueEDUsStatements) InsertQueueEDU( ctx context.Context, txn *sql.Tx, - userID, deviceID string, + eduType string, serverName gomatrixserverlib.ServerName, nid int64, ) error { stmt := sqlutil.TxStmt(txn, s.insertQueueEDUStmt) _, err := stmt.ExecContext( ctx, - userID, // destination user ID - deviceID, // destination device ID + eduType, // the EDU type serverName, // destination server name nid, // JSON blob NID ) return err } -func (s *queueEDUsStatements) SelectQueueEDU( - ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, +func (s *queueEDUsStatements) DeleteQueueEDUs( + ctx context.Context, txn *sql.Tx, + serverName gomatrixserverlib.ServerName, + jsonNIDs []int64, +) error { + stmt := sqlutil.TxStmt(txn, s.deleteQueueEDUStmt) + _, err := stmt.ExecContext(ctx, serverName, pq.Int64Array(jsonNIDs)) + return err +} + +func (s *queueEDUsStatements) SelectQueueEDUs( + ctx context.Context, txn *sql.Tx, + serverName gomatrixserverlib.ServerName, + limit int, ) ([]int64, error) { stmt := sqlutil.TxStmt(txn, s.selectQueueEDUStmt) - rows, err := stmt.QueryContext(ctx) + rows, err := stmt.QueryContext(ctx, serverName, limit) if err != nil { return nil, err } diff --git a/federationsender/storage/postgres/queue_pdus_table.go b/federationsender/storage/postgres/queue_pdus_table.go index 1740487ee..95a3b9eee 100644 --- a/federationsender/storage/postgres/queue_pdus_table.go +++ b/federationsender/storage/postgres/queue_pdus_table.go @@ -18,6 +18,7 @@ import ( "context" "database/sql" + "github.com/lib/pq" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/gomatrixserverlib" @@ -41,8 +42,8 @@ const insertQueuePDUSQL = "" + "INSERT INTO federationsender_queue_pdus (transaction_id, server_name, json_nid)" + " VALUES ($1, $2, $3)" -const deleteQueueTransactionPDUsSQL = "" + - "DELETE FROM federationsender_queue_pdus WHERE server_name = $1 AND transaction_id = $2" +const deleteQueuePDUSQL = "" + + "DELETE FROM federationsender_queue_pdus WHERE server_name = $1 AND json_nid = ANY($2)" const selectQueuePDUNextTransactionIDSQL = "" + "SELECT transaction_id FROM federationsender_queue_pdus" + @@ -69,7 +70,7 @@ const selectQueuePDUServerNamesSQL = "" + type queuePDUsStatements struct { db *sql.DB insertQueuePDUStmt *sql.Stmt - deleteQueueTransactionPDUsStmt *sql.Stmt + deleteQueuePDUsStmt *sql.Stmt selectQueuePDUNextTransactionIDStmt *sql.Stmt selectQueuePDUsByTransactionStmt *sql.Stmt selectQueuePDUReferenceJSONCountStmt *sql.Stmt @@ -88,7 +89,7 @@ func NewPostgresQueuePDUsTable(db *sql.DB) (s *queuePDUsStatements, err error) { if s.insertQueuePDUStmt, err = s.db.Prepare(insertQueuePDUSQL); err != nil { return } - if s.deleteQueueTransactionPDUsStmt, err = s.db.Prepare(deleteQueueTransactionPDUsSQL); err != nil { + if s.deleteQueuePDUsStmt, err = s.db.Prepare(deleteQueuePDUSQL); err != nil { return } if s.selectQueuePDUNextTransactionIDStmt, err = s.db.Prepare(selectQueuePDUNextTransactionIDSQL); err != nil { @@ -126,13 +127,13 @@ func (s *queuePDUsStatements) InsertQueuePDU( return err } -func (s *queuePDUsStatements) DeleteQueuePDUTransaction( +func (s *queuePDUsStatements) DeleteQueuePDUs( ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, - transactionID gomatrixserverlib.TransactionID, + jsonNIDs []int64, ) error { - stmt := sqlutil.TxStmt(txn, s.deleteQueueTransactionPDUsStmt) - _, err := stmt.ExecContext(ctx, serverName, transactionID) + stmt := sqlutil.TxStmt(txn, s.deleteQueuePDUsStmt) + _, err := stmt.ExecContext(ctx, serverName, pq.Int64Array(jsonNIDs)) return err } diff --git a/federationsender/storage/shared/storage.go b/federationsender/storage/shared/storage.go index e5ac3876b..1f9831b94 100644 --- a/federationsender/storage/shared/storage.go +++ b/federationsender/storage/shared/storage.go @@ -21,6 +21,23 @@ type Database struct { FederationSenderRooms tables.FederationSenderRooms } +// An Receipt contains the NIDs of a call to GetNextTransactionPDUs/EDUs. +// We don't actually export the NIDs but we need the caller to be able +// to pass them back so that we can clean up if the transaction sends +// successfully. +type Receipt struct { + nids []int64 +} + +func (e *Receipt) Empty() bool { + return len(e.nids) == 0 +} + +func (e *Receipt) String() string { + j, _ := json.Marshal(e.nids) + return string(j) +} + // UpdateRoom updates the joined hosts for a room and returns what the joined // hosts were before the update, or nil if this was a duplicate message. // This is called when we receive a message from kafka, so we pass in @@ -96,136 +113,12 @@ func (d *Database) GetAllJoinedHosts(ctx context.Context) ([]gomatrixserverlib.S // metadata entries. func (d *Database) StoreJSON( ctx context.Context, js string, -) (int64, error) { +) (*Receipt, error) { nid, err := d.FederationSenderQueueJSON.InsertQueueJSON(ctx, nil, js) if err != nil { - return 0, fmt.Errorf("d.insertQueueJSON: %w", err) + return nil, fmt.Errorf("d.insertQueueJSON: %w", err) } - return nid, nil -} - -// AssociatePDUWithDestination creates an association that the -// destination queues will use to determine which JSON blobs to send -// to which servers. -func (d *Database) AssociatePDUWithDestination( - ctx context.Context, - transactionID gomatrixserverlib.TransactionID, - serverName gomatrixserverlib.ServerName, - nids []int64, -) error { - return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { - for _, nid := range nids { - if err := d.FederationSenderQueuePDUs.InsertQueuePDU( - ctx, // context - txn, // SQL transaction - transactionID, // transaction ID - serverName, // destination server name - nid, // NID from the federationsender_queue_json table - ); err != nil { - return fmt.Errorf("d.insertQueueRetryStmt.ExecContext: %w", err) - } - } - return nil - }) -} - -// GetNextTransactionPDUs retrieves events from the database for -// the next pending transaction, up to the limit specified. -func (d *Database) GetNextTransactionPDUs( - ctx context.Context, - serverName gomatrixserverlib.ServerName, - limit int, -) ( - transactionID gomatrixserverlib.TransactionID, - events []*gomatrixserverlib.HeaderedEvent, - err error, -) { - err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { - transactionID, err = d.FederationSenderQueuePDUs.SelectQueuePDUNextTransactionID(ctx, txn, serverName) - if err != nil { - return fmt.Errorf("d.selectQueueNextTransactionID: %w", err) - } - - if transactionID == "" { - return nil - } - - nids, err := d.FederationSenderQueuePDUs.SelectQueuePDUs(ctx, txn, serverName, transactionID, limit) - if err != nil { - return fmt.Errorf("d.selectQueuePDUs: %w", err) - } - - blobs, err := d.FederationSenderQueueJSON.SelectQueueJSON(ctx, txn, nids) - if err != nil { - return fmt.Errorf("d.selectJSON: %w", err) - } - - for _, blob := range blobs { - var event gomatrixserverlib.HeaderedEvent - if err := json.Unmarshal(blob, &event); err != nil { - return fmt.Errorf("json.Unmarshal: %w", err) - } - events = append(events, &event) - } - - return nil - }) - return -} - -// CleanTransactionPDUs cleans up all associated events for a -// given transaction. This is done when the transaction was sent -// successfully. -func (d *Database) CleanTransactionPDUs( - ctx context.Context, - serverName gomatrixserverlib.ServerName, - transactionID gomatrixserverlib.TransactionID, -) error { - return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { - nids, err := d.FederationSenderQueuePDUs.SelectQueuePDUs(ctx, txn, serverName, transactionID, 50) - if err != nil { - return fmt.Errorf("d.selectQueuePDUs: %w", err) - } - - if err = d.FederationSenderQueuePDUs.DeleteQueuePDUTransaction(ctx, txn, serverName, transactionID); err != nil { - return fmt.Errorf("d.deleteQueueTransaction: %w", err) - } - - var count int64 - var deleteNIDs []int64 - for _, nid := range nids { - count, err = d.FederationSenderQueuePDUs.SelectQueuePDUReferenceJSONCount(ctx, txn, nid) - if err != nil { - return fmt.Errorf("d.selectQueueReferenceJSONCount: %w", err) - } - if count == 0 { - deleteNIDs = append(deleteNIDs, nid) - } - } - - if len(deleteNIDs) > 0 { - if err = d.FederationSenderQueueJSON.DeleteQueueJSON(ctx, txn, deleteNIDs); err != nil { - return fmt.Errorf("d.deleteQueueJSON: %w", err) - } - } - - return nil - }) -} - -// GetPendingPDUCount returns the number of PDUs waiting to be -// sent for a given servername. -func (d *Database) GetPendingPDUCount( - ctx context.Context, - serverName gomatrixserverlib.ServerName, -) (int64, error) { - return d.FederationSenderQueuePDUs.SelectQueuePDUCount(ctx, nil, serverName) -} - -// GetPendingServerNames returns the server names that have PDUs -// waiting to be sent. -func (d *Database) GetPendingPDUServerNames( - ctx context.Context, -) ([]gomatrixserverlib.ServerName, error) { - return d.FederationSenderQueuePDUs.SelectQueuePDUServerNames(ctx, nil) + return &Receipt{ + nids: []int64{nid}, + }, nil } diff --git a/federationsender/storage/shared/storage_edus.go b/federationsender/storage/shared/storage_edus.go new file mode 100644 index 000000000..a63cacf2d --- /dev/null +++ b/federationsender/storage/shared/storage_edus.go @@ -0,0 +1,129 @@ +package shared + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/matrix-org/dendrite/internal/sqlutil" + "github.com/matrix-org/gomatrixserverlib" +) + +// AssociateEDUWithDestination creates an association that the +// destination queues will use to determine which JSON blobs to send +// to which servers. +func (d *Database) AssociateEDUWithDestination( + ctx context.Context, + serverName gomatrixserverlib.ServerName, + receipt *Receipt, +) error { + return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { + for _, nid := range receipt.nids { + if err := d.FederationSenderQueueEDUs.InsertQueueEDU( + ctx, // context + txn, // SQL transaction + "", // TODO: EDU type for coalescing + serverName, // destination server name + nid, // NID from the federationsender_queue_json table + ); err != nil { + return fmt.Errorf("InsertQueueEDU: %w", err) + } + } + return nil + }) +} + +// GetNextTransactionEDUs retrieves events from the database for +// the next pending transaction, up to the limit specified. +func (d *Database) GetNextTransactionEDUs( + ctx context.Context, + serverName gomatrixserverlib.ServerName, + limit int, +) ( + edus []*gomatrixserverlib.EDU, + receipt *Receipt, + err error, +) { + err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { + nids, err := d.FederationSenderQueueEDUs.SelectQueueEDUs(ctx, txn, serverName, limit) + if err != nil { + return fmt.Errorf("SelectQueueEDUs: %w", err) + } + + receipt = &Receipt{ + nids: nids, + } + + blobs, err := d.FederationSenderQueueJSON.SelectQueueJSON(ctx, txn, nids) + if err != nil { + return fmt.Errorf("SelectQueueJSON: %w", err) + } + + for _, blob := range blobs { + var event gomatrixserverlib.EDU + if err := json.Unmarshal(blob, &event); err != nil { + return fmt.Errorf("json.Unmarshal: %w", err) + } + edus = append(edus, &event) + } + + return nil + }) + return +} + +// CleanEDUs cleans up all specified EDUs. This is done when a +// transaction was sent successfully. +func (d *Database) CleanEDUs( + ctx context.Context, + serverName gomatrixserverlib.ServerName, + receipt *Receipt, +) error { + if receipt == nil { + return errors.New("expected receipt") + } + + return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { + if err := d.FederationSenderQueueEDUs.DeleteQueueEDUs(ctx, txn, serverName, receipt.nids); err != nil { + return err + } + + var deleteNIDs []int64 + for _, nid := range receipt.nids { + count, err := d.FederationSenderQueueEDUs.SelectQueueEDUReferenceJSONCount(ctx, txn, nid) + if err != nil { + return fmt.Errorf("SelectQueueEDUReferenceJSONCount: %w", err) + } + if count == 0 { + deleteNIDs = append(deleteNIDs, nid) + } + } + + if len(deleteNIDs) > 0 { + if err := d.FederationSenderQueueJSON.DeleteQueueJSON(ctx, txn, deleteNIDs); err != nil { + return fmt.Errorf("DeleteQueueJSON: %w", err) + } + } + + return nil + }) +} + +// GetPendingEDUCount returns the number of EDUs waiting to be +// sent for a given servername. +func (d *Database) GetPendingEDUCount( + ctx context.Context, + serverName gomatrixserverlib.ServerName, +) (int64, error) { + return d.FederationSenderQueueEDUs.SelectQueueEDUCount(ctx, nil, serverName) +} + +// GetPendingServerNames returns the server names that have EDUs +// waiting to be sent. +func (d *Database) GetPendingEDUServerNames( + ctx context.Context, +) ([]gomatrixserverlib.ServerName, error) { + return d.FederationSenderQueueEDUs.SelectQueueEDUServerNames(ctx, nil) +} diff --git a/federationsender/storage/shared/storage_pdus.go b/federationsender/storage/shared/storage_pdus.go new file mode 100644 index 000000000..ea1382859 --- /dev/null +++ b/federationsender/storage/shared/storage_pdus.go @@ -0,0 +1,141 @@ +package shared + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/matrix-org/dendrite/internal/sqlutil" + "github.com/matrix-org/gomatrixserverlib" +) + +// AssociatePDUWithDestination creates an association that the +// destination queues will use to determine which JSON blobs to send +// to which servers. +func (d *Database) AssociatePDUWithDestination( + ctx context.Context, + transactionID gomatrixserverlib.TransactionID, + serverName gomatrixserverlib.ServerName, + receipt *Receipt, +) error { + return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { + for _, nid := range receipt.nids { + if err := d.FederationSenderQueuePDUs.InsertQueuePDU( + ctx, // context + txn, // SQL transaction + transactionID, // transaction ID + serverName, // destination server name + nid, // NID from the federationsender_queue_json table + ); err != nil { + return fmt.Errorf("InsertQueuePDU: %w", err) + } + } + return nil + }) +} + +// GetNextTransactionPDUs retrieves events from the database for +// the next pending transaction, up to the limit specified. +func (d *Database) GetNextTransactionPDUs( + ctx context.Context, + serverName gomatrixserverlib.ServerName, + limit int, +) ( + transactionID gomatrixserverlib.TransactionID, + events []*gomatrixserverlib.HeaderedEvent, + receipt *Receipt, + err error, +) { + err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { + transactionID, err = d.FederationSenderQueuePDUs.SelectQueuePDUNextTransactionID(ctx, txn, serverName) + if err != nil { + return fmt.Errorf("SelectQueuePDUNextTransactionID: %w", err) + } + + if transactionID == "" { + return nil + } + + nids, err := d.FederationSenderQueuePDUs.SelectQueuePDUs(ctx, txn, serverName, transactionID, limit) + if err != nil { + return fmt.Errorf("SelectQueuePDUs: %w", err) + } + + receipt = &Receipt{ + nids: nids, + } + + blobs, err := d.FederationSenderQueueJSON.SelectQueueJSON(ctx, txn, nids) + if err != nil { + return fmt.Errorf("SelectQueueJSON: %w", err) + } + + for _, blob := range blobs { + var event gomatrixserverlib.HeaderedEvent + if err := json.Unmarshal(blob, &event); err != nil { + return fmt.Errorf("json.Unmarshal: %w", err) + } + events = append(events, &event) + } + + return nil + }) + return +} + +// CleanTransactionPDUs cleans up all associated events for a +// given transaction. This is done when the transaction was sent +// successfully. +func (d *Database) CleanPDUs( + ctx context.Context, + serverName gomatrixserverlib.ServerName, + receipt *Receipt, +) error { + if receipt == nil { + return errors.New("expected receipt") + } + + return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error { + if err := d.FederationSenderQueuePDUs.DeleteQueuePDUs(ctx, txn, serverName, receipt.nids); err != nil { + return err + } + + var deleteNIDs []int64 + for _, nid := range receipt.nids { + count, err := d.FederationSenderQueuePDUs.SelectQueuePDUReferenceJSONCount(ctx, txn, nid) + if err != nil { + return fmt.Errorf("SelectQueuePDUReferenceJSONCount: %w", err) + } + if count == 0 { + deleteNIDs = append(deleteNIDs, nid) + } + } + + if len(deleteNIDs) > 0 { + if err := d.FederationSenderQueueJSON.DeleteQueueJSON(ctx, txn, deleteNIDs); err != nil { + return fmt.Errorf("DeleteQueueJSON: %w", err) + } + } + + return nil + }) +} + +// GetPendingPDUCount returns the number of PDUs waiting to be +// sent for a given servername. +func (d *Database) GetPendingPDUCount( + ctx context.Context, + serverName gomatrixserverlib.ServerName, +) (int64, error) { + return d.FederationSenderQueuePDUs.SelectQueuePDUCount(ctx, nil, serverName) +} + +// GetPendingServerNames returns the server names that have PDUs +// waiting to be sent. +func (d *Database) GetPendingPDUServerNames( + ctx context.Context, +) ([]gomatrixserverlib.ServerName, error) { + return d.FederationSenderQueuePDUs.SelectQueuePDUServerNames(ctx, nil) +} diff --git a/federationsender/storage/sqlite3/queue_edus_table.go b/federationsender/storage/sqlite3/queue_edus_table.go index aefe36e85..000ae104c 100644 --- a/federationsender/storage/sqlite3/queue_edus_table.go +++ b/federationsender/storage/sqlite3/queue_edus_table.go @@ -18,6 +18,8 @@ package sqlite3 import ( "context" "database/sql" + "fmt" + "strings" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/sqlutil" @@ -42,9 +44,13 @@ const insertQueueEDUSQL = "" + "INSERT INTO federationsender_queue_edus (edu_type, server_name, json_nid)" + " VALUES ($1, $2, $3)" +const deleteQueueEDUsSQL = "" + + "DELETE FROM federationsender_queue_edus WHERE server_name = $1 AND json_nid IN ($2)" + const selectQueueEDUSQL = "" + "SELECT json_nid FROM federationsender_queue_edus" + - " WHERE server_name = $1" + " WHERE server_name = $1" + + " LIMIT $2" const selectQueueEDUReferenceJSONCountSQL = "" + "SELECT COUNT(*) FROM federationsender_queue_edus" + @@ -97,7 +103,7 @@ func NewSQLiteQueueEDUsTable(db *sql.DB) (s *queueEDUsStatements, err error) { func (s *queueEDUsStatements) InsertQueueEDU( ctx context.Context, txn *sql.Tx, - userID, deviceID string, + eduType string, serverName gomatrixserverlib.ServerName, nid int64, ) error { @@ -105,8 +111,7 @@ func (s *queueEDUsStatements) InsertQueueEDU( stmt := sqlutil.TxStmt(txn, s.insertQueueEDUStmt) _, err := stmt.ExecContext( ctx, - userID, // destination user ID - deviceID, // destination device ID + eduType, // the EDU type serverName, // destination server name nid, // JSON blob NID ) @@ -114,11 +119,38 @@ func (s *queueEDUsStatements) InsertQueueEDU( }) } -func (s *queueEDUsStatements) SelectQueueEDU( - ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, +func (s *queueEDUsStatements) DeleteQueueEDUs( + ctx context.Context, txn *sql.Tx, + serverName gomatrixserverlib.ServerName, + jsonNIDs []int64, +) error { + deleteSQL := strings.Replace(deleteQueueEDUsSQL, "($2)", sqlutil.QueryVariadicOffset(len(jsonNIDs), 1), 1) + fmt.Println(deleteSQL) + deleteStmt, err := txn.Prepare(deleteSQL) + if err != nil { + return fmt.Errorf("s.deleteQueueJSON s.db.Prepare: %w", err) + } + + params := make([]interface{}, len(jsonNIDs)+1) + params[0] = serverName + for k, v := range jsonNIDs { + params[k+1] = v + } + + return s.writer.Do(s.db, txn, func(txn *sql.Tx) error { + stmt := sqlutil.TxStmt(txn, deleteStmt) + _, err := stmt.ExecContext(ctx, params...) + return err + }) +} + +func (s *queueEDUsStatements) SelectQueueEDUs( + ctx context.Context, txn *sql.Tx, + serverName gomatrixserverlib.ServerName, + limit int, ) ([]int64, error) { stmt := sqlutil.TxStmt(txn, s.selectQueueEDUStmt) - rows, err := stmt.QueryContext(ctx) + rows, err := stmt.QueryContext(ctx, serverName, limit) if err != nil { return nil, err } diff --git a/federationsender/storage/sqlite3/queue_pdus_table.go b/federationsender/storage/sqlite3/queue_pdus_table.go index 1fc5680c3..9d8eaab66 100644 --- a/federationsender/storage/sqlite3/queue_pdus_table.go +++ b/federationsender/storage/sqlite3/queue_pdus_table.go @@ -18,6 +18,8 @@ package sqlite3 import ( "context" "database/sql" + "fmt" + "strings" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/sqlutil" @@ -42,8 +44,8 @@ const insertQueuePDUSQL = "" + "INSERT INTO federationsender_queue_pdus (transaction_id, server_name, json_nid)" + " VALUES ($1, $2, $3)" -const deleteQueueTransactionPDUsSQL = "" + - "DELETE FROM federationsender_queue_pdus WHERE server_name = $1 AND transaction_id = $2" +const deleteQueuePDUsSQL = "" + + "DELETE FROM federationsender_queue_pdus WHERE server_name = $1 AND json_nid IN ($2)" const selectQueueNextTransactionIDSQL = "" + "SELECT transaction_id FROM federationsender_queue_pdus" + @@ -71,12 +73,12 @@ type queuePDUsStatements struct { db *sql.DB writer *sqlutil.TransactionWriter insertQueuePDUStmt *sql.Stmt - deleteQueueTransactionPDUsStmt *sql.Stmt selectQueueNextTransactionIDStmt *sql.Stmt selectQueuePDUsByTransactionStmt *sql.Stmt selectQueueReferenceJSONCountStmt *sql.Stmt selectQueuePDUsCountStmt *sql.Stmt selectQueueServerNamesStmt *sql.Stmt + // deleteQueuePDUsStmt *sql.Stmt - prepared at runtime due to variadic } func NewSQLiteQueuePDUsTable(db *sql.DB) (s *queuePDUsStatements, err error) { @@ -91,9 +93,9 @@ func NewSQLiteQueuePDUsTable(db *sql.DB) (s *queuePDUsStatements, err error) { if s.insertQueuePDUStmt, err = db.Prepare(insertQueuePDUSQL); err != nil { return } - if s.deleteQueueTransactionPDUsStmt, err = db.Prepare(deleteQueueTransactionPDUsSQL); err != nil { - return - } + //if s.deleteQueuePDUsStmt, err = db.Prepare(deleteQueuePDUsSQL); err != nil { + // return + //} if s.selectQueueNextTransactionIDStmt, err = db.Prepare(selectQueueNextTransactionIDSQL); err != nil { return } @@ -131,14 +133,27 @@ func (s *queuePDUsStatements) InsertQueuePDU( }) } -func (s *queuePDUsStatements) DeleteQueuePDUTransaction( +func (s *queuePDUsStatements) DeleteQueuePDUs( ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, - transactionID gomatrixserverlib.TransactionID, + jsonNIDs []int64, ) error { + deleteSQL := strings.Replace(deleteQueuePDUsSQL, "($2)", sqlutil.QueryVariadicOffset(len(jsonNIDs), 1), 1) + fmt.Println(deleteSQL) + deleteStmt, err := txn.Prepare(deleteSQL) + if err != nil { + return fmt.Errorf("s.deleteQueueJSON s.db.Prepare: %w", err) + } + + params := make([]interface{}, len(jsonNIDs)+1) + params[0] = serverName + for k, v := range jsonNIDs { + params[k+1] = v + } + return s.writer.Do(s.db, txn, func(txn *sql.Tx) error { - stmt := sqlutil.TxStmt(txn, s.deleteQueueTransactionPDUsStmt) - _, err := stmt.ExecContext(ctx, serverName, transactionID) + stmt := sqlutil.TxStmt(txn, deleteStmt) + _, err := stmt.ExecContext(ctx, params...) return err }) } diff --git a/federationsender/storage/tables/interface.go b/federationsender/storage/tables/interface.go index e4155f0c2..fed644de7 100644 --- a/federationsender/storage/tables/interface.go +++ b/federationsender/storage/tables/interface.go @@ -10,7 +10,7 @@ import ( type FederationSenderQueuePDUs interface { InsertQueuePDU(ctx context.Context, txn *sql.Tx, transactionID gomatrixserverlib.TransactionID, serverName gomatrixserverlib.ServerName, nid int64) error - DeleteQueuePDUTransaction(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, transactionID gomatrixserverlib.TransactionID) error + DeleteQueuePDUs(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, jsonNIDs []int64) error SelectQueuePDUNextTransactionID(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) (gomatrixserverlib.TransactionID, error) SelectQueuePDUReferenceJSONCount(ctx context.Context, txn *sql.Tx, jsonNID int64) (int64, error) SelectQueuePDUCount(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) (int64, error) @@ -19,8 +19,9 @@ type FederationSenderQueuePDUs interface { } type FederationSenderQueueEDUs interface { - InsertQueueEDU(ctx context.Context, txn *sql.Tx, userID, deviceID string, serverName gomatrixserverlib.ServerName, nid int64) error - SelectQueueEDU(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) ([]int64, error) + InsertQueueEDU(ctx context.Context, txn *sql.Tx, eduType string, serverName gomatrixserverlib.ServerName, nid int64) error + DeleteQueueEDUs(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, jsonNIDs []int64) error + SelectQueueEDUs(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, limit int) ([]int64, error) SelectQueueEDUReferenceJSONCount(ctx context.Context, txn *sql.Tx, jsonNID int64) (int64, error) SelectQueueEDUCount(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) (int64, error) SelectQueueEDUServerNames(ctx context.Context, txn *sql.Tx) ([]gomatrixserverlib.ServerName, error)