diff --git a/federationapi/queue/destinationqueue.go b/federationapi/queue/destinationqueue.go index 00e02b2d9..768ed1f2b 100644 --- a/federationapi/queue/destinationqueue.go +++ b/federationapi/queue/destinationqueue.go @@ -21,16 +21,17 @@ import ( "sync" "time" + "github.com/matrix-org/gomatrix" + "github.com/matrix-org/gomatrixserverlib" + "github.com/sirupsen/logrus" + "go.uber.org/atomic" + fedapi "github.com/matrix-org/dendrite/federationapi/api" "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 ( @@ -541,6 +542,8 @@ func (oq *destinationQueue) handleTransactionSuccess(pduCount int, eduCount int) // the pending events and EDUs, and wipe our transaction ID. oq.statistics.Success() oq.pendingMutex.Lock() + defer oq.pendingMutex.Unlock() + for i := range oq.pendingPDUs[:pduCount] { oq.pendingPDUs[i] = nil } @@ -549,7 +552,6 @@ func (oq *destinationQueue) handleTransactionSuccess(pduCount int, eduCount int) } oq.pendingPDUs = oq.pendingPDUs[pduCount:] oq.pendingEDUs = oq.pendingEDUs[eduCount:] - oq.pendingMutex.Unlock() if len(oq.pendingPDUs) > 0 || len(oq.pendingEDUs) > 0 { select { diff --git a/federationapi/queue/queue_test.go b/federationapi/queue/queue_test.go index 6da9e6b30..40419b91f 100644 --- a/federationapi/queue/queue_test.go +++ b/federationapi/queue/queue_test.go @@ -1004,9 +1004,12 @@ func TestQueueInteractsWithRealDatabasePDUAndEDU(t *testing.T) { err := queues.SendEvent(ev, "localhost", []gomatrixserverlib.ServerName{destination}) assert.NoError(t, err) + // NOTE : The server can be blacklisted before this, so manually inject the event + // into the database. edu := mustCreateEDU(t) - errEDU := queues.SendEDU(edu, "localhost", []gomatrixserverlib.ServerName{destination}) - assert.NoError(t, errEDU) + ephemeralJSON, _ := json.Marshal(edu) + nid, _ := db.StoreJSON(pc.Context(), string(ephemeralJSON)) + db.AssociateEDUWithDestination(pc.Context(), destination, nid, edu.Type, nil) checkBlacklisted := func(log poll.LogT) poll.Result { if fc.txCount.Load() == failuresUntilBlacklist { diff --git a/sytest-blacklist b/sytest-blacklist index fe48fb791..14edf398a 100644 --- a/sytest-blacklist +++ b/sytest-blacklist @@ -22,10 +22,6 @@ Forgotten room messages cannot be paginated Local device key changes get to remote servers with correct prev_id -# Flakey - -Local device key changes appear in /keys/changes - # we don't support groups Remove group category