mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-21 04:53:14 -06:00
Code Issues Packages Projects Releases Wiki Activity

Fix FederationAPI not processing purges; start SyncAPI/FedAPI consumer

Browse source
This commit is contained in:
Till Faelligen 2022-12-21 16:49:10 +01:00
parent c10f563510
commit 23d1002142
No known key found for this signature in database
GPG key ID: ACCDC9606D472758
2 changed files with 19 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
federationapi/consumers/roomserver.go
View file

@ -91,8 +91,10 @@ func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Ms
msg := msgs[0] // Guaranteed to exist if onMessage is called msg := msgs[0] // Guaranteed to exist if onMessage is called
receivedType := api.OutputType(msg.Header.Get(jetstream.RoomEventType)) receivedType := api.OutputType(msg.Header.Get(jetstream.RoomEventType))
// Only handle events we care about // Only handle events we care about, avoids unneeded unmarshalling
if receivedType != api.OutputTypeNewRoomEvent && receivedType != api.OutputTypeNewInboundPeek { switch receivedType {
case api.OutputTypeNewRoomEvent, api.OutputTypeNewInboundPeek, api.OutputTypePurgeRoom:
default:
return true return true
} }

16
roomserver/roomserver_test.go
View file

@ -7,6 +7,10 @@ import (
"time" "time"
"github.com/gorilla/mux" "github.com/gorilla/mux"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/syncapi"
"github.com/matrix-org/dendrite/userapi"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/httputil" "github.com/matrix-org/dendrite/internal/httputil"
@ -160,9 +164,16 @@ func TestPurgeRoom(t *testing.T) {
base, db, close := mustCreateDatabase(t, dbType) base, db, close := mustCreateDatabase(t, dbType)
defer close() defer close()
fedClient := base.CreateFederationClient()
rsAPI := roomserver.NewInternalAPI(base) rsAPI := roomserver.NewInternalAPI(base)
// SetFederationAPI starts the room event input consumer keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fedClient, rsAPI)
userAPI := userapi.NewInternalAPI(base, &base.Cfg.UserAPI, nil, keyAPI, rsAPI, nil)
// this starts the JetStream consumers
syncapi.AddPublicRoutes(base, userAPI, rsAPI, keyAPI)
federationapi.NewInternalAPI(base, fedClient, rsAPI, base.Caches, nil, true)
rsAPI.SetFederationAPI(nil, nil) rsAPI.SetFederationAPI(nil, nil)
// Create the room // Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil { if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err) t.Fatalf("failed to send events: %v", err)
@ -237,6 +248,9 @@ func TestPurgeRoom(t *testing.T) {
if err = rsAPI.PerformAdminPurgeRoom(ctx, &api.PerformAdminPurgeRoomRequest{RoomID: room.ID}, purgeResp); err != nil { if err = rsAPI.PerformAdminPurgeRoom(ctx, &api.PerformAdminPurgeRoomRequest{RoomID: room.ID}, purgeResp); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// TODO: Find a better solution, e.g. "hook" into the JetStream stream.
// Gives the SyncAPI and FederationAPI some time to delete their entries
time.Sleep(time.Millisecond * 100)
roomInfo, err = db.RoomInfo(ctx, room.ID) roomInfo, err = db.RoomInfo(ctx, room.ID)
if err != nil { if err != nil {