From c4ec66e892353ad086b1ce6cbcab5be201cd6f8e Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Thu, 30 Apr 2020 16:42:14 +0100 Subject: [PATCH] Add tests around syncing --- syncapi/storage/interface.go | 3 +- syncapi/storage/storage_test.go | 291 ++++++++++++++++++++++++++++++++ 2 files changed, 293 insertions(+), 1 deletion(-) create mode 100644 syncapi/storage/storage_test.go diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go index 4b789e56d..bd9504db8 100644 --- a/syncapi/storage/interface.go +++ b/syncapi/storage/interface.go @@ -39,7 +39,8 @@ type Database interface { // WriteEvent into the database. It is not safe to call this function from multiple goroutines, as it would create races // when generating the sync stream position for this event. Returns the sync stream position for the inserted event. // Returns an error if there was a problem inserting this event. - WriteEvent(context.Context, *gomatrixserverlib.HeaderedEvent, []gomatrixserverlib.HeaderedEvent, []string, []string, *api.TransactionID, bool) (types.StreamPosition, error) + WriteEvent(ctx context.Context, ev *gomatrixserverlib.HeaderedEvent, addStateEvents []gomatrixserverlib.HeaderedEvent, + addStateEventIDs []string, removeStateEventIDs []string, transactionID *api.TransactionID, excludeFromSync bool) (types.StreamPosition, error) // GetStateEvent returns the Matrix state event of a given type for a given room with a given state key // If no event could be found, returns nil // If there was an issue during the retrieval, returns an error diff --git a/syncapi/storage/storage_test.go b/syncapi/storage/storage_test.go new file mode 100644 index 000000000..35d7abafc --- /dev/null +++ b/syncapi/storage/storage_test.go @@ -0,0 +1,291 @@ +package storage_test + +import ( + "context" + "crypto/ed25519" + "fmt" + "testing" + "time" + + "github.com/matrix-org/dendrite/clientapi/auth/authtypes" + "github.com/matrix-org/dendrite/syncapi/storage" + "github.com/matrix-org/dendrite/syncapi/storage/sqlite3" + "github.com/matrix-org/dendrite/syncapi/types" + "github.com/matrix-org/gomatrixserverlib" +) + +var ( + ctx = context.Background() + emptyStateKey = "" + testOrigin = gomatrixserverlib.ServerName("hollow.knight") + testRoomID = fmt.Sprintf("!hallownest:%s", testOrigin) + testUserIDA = fmt.Sprintf("@hornet:%s", testOrigin) + testUserIDB = fmt.Sprintf("@paleking:%s", testOrigin) + testUserDeviceA = authtypes.Device{ + UserID: testUserIDA, + ID: "device_id_A", + DisplayName: "Device A", + } + testUserDeviceB = authtypes.Device{ + UserID: testUserIDB, + ID: "device_id_B", + DisplayName: "Device B", + } + nullDevice = authtypes.Device{} + testRoomVersion = gomatrixserverlib.RoomVersionV4 + testKeyID = gomatrixserverlib.KeyID("ed25519:storage_test") + testPrivateKey = ed25519.NewKeyFromSeed([]byte{ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, + 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, + }) +) + +func MustCreateEvent(t *testing.T, roomID string, prevs []gomatrixserverlib.HeaderedEvent, b *gomatrixserverlib.EventBuilder) gomatrixserverlib.HeaderedEvent { + b.RoomID = roomID + if prevs != nil { + prevIDs := make([]string, len(prevs)) + for i := range prevs { + prevIDs[i] = prevs[i].EventID() + } + b.PrevEvents = prevIDs + } + e, err := b.Build(time.Now(), testOrigin, testKeyID, testPrivateKey, testRoomVersion) + if err != nil { + t.Fatalf("failed to build event: %s", err) + } + return e.Headered(testRoomVersion) +} + +func MustCreateDatabase(t *testing.T) storage.Database { + db, err := sqlite3.NewSyncServerDatasource("file::memory:") + if err != nil { + t.Fatalf("NewSyncServerDatasource returned %s", err) + } + return db +} + +// Create a list of events which include a create event, join event and some messages. +func SimpleRoom(t *testing.T, roomID, userA, userB string) (msgs []gomatrixserverlib.HeaderedEvent, state []gomatrixserverlib.HeaderedEvent) { + var events []gomatrixserverlib.HeaderedEvent + events = append(events, MustCreateEvent(t, roomID, nil, &gomatrixserverlib.EventBuilder{ + Content: []byte(fmt.Sprintf(`{"room_version":"4","creator":"%s"}`, userA)), + Type: "m.room.create", + StateKey: &emptyStateKey, + Sender: userA, + })) + state = append(state, events[len(events)-1]) + events = append(events, MustCreateEvent(t, roomID, []gomatrixserverlib.HeaderedEvent{events[len(events)-1]}, &gomatrixserverlib.EventBuilder{ + Content: []byte(fmt.Sprintf(`{"membership":"join"}`)), + Type: "m.room.member", + StateKey: &userA, + Sender: userA, + })) + state = append(state, events[len(events)-1]) + for i := 0; i < 10; i++ { + events = append(events, MustCreateEvent(t, roomID, []gomatrixserverlib.HeaderedEvent{events[len(events)-1]}, &gomatrixserverlib.EventBuilder{ + Content: []byte(fmt.Sprintf(`{"body":"Message A %d"}`, i+1)), + Type: "m.room.message", + Sender: userA, + })) + } + events = append(events, MustCreateEvent(t, roomID, []gomatrixserverlib.HeaderedEvent{events[len(events)-1]}, &gomatrixserverlib.EventBuilder{ + Content: []byte(fmt.Sprintf(`{"membership":"join"}`)), + Type: "m.room.member", + StateKey: &userB, + Sender: userB, + })) + state = append(state, events[len(events)-1]) + for i := 0; i < 10; i++ { + events = append(events, MustCreateEvent(t, roomID, []gomatrixserverlib.HeaderedEvent{events[len(events)-1]}, &gomatrixserverlib.EventBuilder{ + Content: []byte(fmt.Sprintf(`{"body":"Message B %d"}`, i+1)), + Type: "m.room.message", + Sender: userB, + })) + } + + return events, state +} + +func MustWriteEvents(t *testing.T, db storage.Database, events []gomatrixserverlib.HeaderedEvent) (positions []types.StreamPosition) { + for _, ev := range events { + var addStateEvents []gomatrixserverlib.HeaderedEvent + var addStateEventIDs []string + var removeStateEventIDs []string + if ev.StateKey() != nil { + addStateEvents = append(addStateEvents, ev) + addStateEventIDs = append(addStateEventIDs, ev.EventID()) + } + pos, err := db.WriteEvent(ctx, &ev, addStateEvents, addStateEventIDs, removeStateEventIDs, nil, false) + if err != nil { + t.Fatalf("WriteEvent failed: %s", err) + } + positions = append(positions, pos) + } + return +} + +func TestWriteEvents(t *testing.T) { + t.Parallel() + db := MustCreateDatabase(t) + events, _ := SimpleRoom(t, testRoomID, testUserIDA, testUserIDB) + MustWriteEvents(t, db, events) +} + +// These tests assert basic functionality of the IncrementalSync and CompleteSync functions. +func TestSyncResponse(t *testing.T) { + t.Parallel() + db := MustCreateDatabase(t) + events, state := SimpleRoom(t, testRoomID, testUserIDA, testUserIDB) + positions := MustWriteEvents(t, db, events) + latest, err := db.SyncPosition(ctx) + if err != nil { + t.Fatalf("failed to get SyncPosition: %s", err) + } + + testCases := []struct { + Name string + DoSync func() (*types.Response, error) + WantTimeline []gomatrixserverlib.HeaderedEvent + WantState []gomatrixserverlib.HeaderedEvent + }{ + // The purpose of this test is to make sure that incremental syncs are including up to the latest events. + // It's a basic sanity test that sync works. It creates a `since` token that is on the penultimate event. + // It makes sure the response includes the final event. + { + Name: "IncrementalSync penultimate", + DoSync: func() (*types.Response, error) { + from := types.NewPaginationTokenFromTypeAndPosition( // pretend we are at the penultimate event + types.PaginationTokenTypeStream, positions[len(positions)-2], types.StreamPosition(0), + ) + return db.IncrementalSync(ctx, testUserDeviceA, *from, latest, 5, false) + }, + WantTimeline: events[len(events)-1:], + }, + // The purpose of this test is to check that passing a `numRecentEventsPerRoom` correctly limits the + // number of returned events. This is critical for big rooms hence the test here. + { + Name: "IncrementalSync limited", + DoSync: func() (*types.Response, error) { + from := types.NewPaginationTokenFromTypeAndPosition( // pretend we are 10 events behind + types.PaginationTokenTypeStream, positions[len(positions)-11], types.StreamPosition(0), + ) + // limit is set to 5 + return db.IncrementalSync(ctx, testUserDeviceA, *from, latest, 5, false) + }, + // want the last 5 events, NOT the last 10. + WantTimeline: events[len(events)-5:], + }, + // The purpose of this test is to check that CompleteSync returns all the current state as well as + // honouring the `numRecentEventsPerRoom` value + { + Name: "CompleteSync limited", + DoSync: func() (*types.Response, error) { + // limit set to 5 + return db.CompleteSync(ctx, testUserIDA, 5) + }, + // want the last 5 events, NOT the last 10. + WantTimeline: events[len(events)-5:], + // want all state for the room + WantState: state, + }, + // The purpose of this test is to check that CompleteSync can return everything with a high enough + // `numRecentEventsPerRoom`. + { + Name: "CompleteSync", + DoSync: func() (*types.Response, error) { + return db.CompleteSync(ctx, testUserIDA, len(events)+1) + }, + WantTimeline: events, + // We want no state at all as that field in /sync is the delta between the token (beginning of time) + // and the START of the timeline. + }, + } + + for _, tc := range testCases { + t.Run(tc.Name, func(st *testing.T) { + res, err := tc.DoSync() + if err != nil { + st.Fatalf("failed to do sync: %s", err) + } + next := types.NewPaginationTokenFromTypeAndPosition(types.PaginationTokenTypeStream, latest.PDUPosition, latest.EDUTypingPosition) + if res.NextBatch != next.String() { + st.Errorf("NextBatch got %s want %s", res.NextBatch, next.String()) + } + roomRes, ok := res.Rooms.Join[testRoomID] + if !ok { + st.Fatalf("IncrementalSync response missing room %s - response: %+v", testRoomID, res) + } + assertEventsEqual(st, "state for "+testRoomID, false, roomRes.State.Events, tc.WantState) + assertEventsEqual(st, "timeline for "+testRoomID, false, roomRes.Timeline.Events, tc.WantTimeline) + }) + } +} + +// The purpose of this test is to ensure that backfill does indeed go backwards. +func TestGetEventsInRange(t *testing.T) { + t.Parallel() + db := MustCreateDatabase(t) + events, _ := SimpleRoom(t, testRoomID, testUserIDA, testUserIDB) + MustWriteEvents(t, db, events) + latest, err := db.SyncPosition(ctx) + if err != nil { + t.Fatalf("failed to get SyncPosition: %s", err) + } + // head towards the beginning of time + to := types.NewPaginationTokenFromTypeAndPosition(types.PaginationTokenTypeTopology, 0, 0) + + // backpaginate 5 messages starting at the latest position. + paginatedEvents, err := db.GetEventsInRange(ctx, &latest, to, testRoomID, 5, true) + if err != nil { + t.Fatalf("GetEventsInRange returned an error: %s", err) + } + gots := gomatrixserverlib.HeaderedToClientEvents(db.StreamEventsToEvents(&testUserDeviceA, paginatedEvents), gomatrixserverlib.FormatAll) + assertEventsEqual(t, "", true, gots, reversed(events[len(events)-5:])) +} + +func assertEventsEqual(t *testing.T, msg string, checkRoomID bool, gots []gomatrixserverlib.ClientEvent, wants []gomatrixserverlib.HeaderedEvent) { + if len(gots) != len(wants) { + t.Fatalf("%s response returned %d events, want %d", msg, len(gots), len(wants)) + } + for i := range gots { + g := gots[i] + w := wants[i] + if g.EventID != w.EventID() { + t.Errorf("%s event[%d] event_id mismatch: got %s want %s", msg, i, g.EventID, w.EventID()) + } + if g.Sender != w.Sender() { + t.Errorf("%s event[%d] sender mismatch: got %s want %s", msg, i, g.Sender, w.Sender()) + } + if checkRoomID && g.RoomID != w.RoomID() { + t.Errorf("%s event[%d] room_id mismatch: got %s want %s", msg, i, g.RoomID, w.RoomID()) + } + if g.Type != w.Type() { + t.Errorf("%s event[%d] event type mismatch: got %s want %s", msg, i, g.Type, w.Type()) + } + if g.OriginServerTS != w.OriginServerTS() { + t.Errorf("%s event[%d] origin_server_ts mismatch: got %v want %v", msg, i, g.OriginServerTS, w.OriginServerTS()) + } + if string(g.Content) != string(w.Content()) { + t.Errorf("%s event[%d] content mismatch: got %s want %s", msg, i, string(g.Content), string(w.Content())) + } + if string(g.Unsigned) != string(w.Unsigned()) { + t.Errorf("%s event[%d] unsigned mismatch: got %s want %s", msg, i, string(g.Unsigned), string(w.Unsigned())) + } + if (g.StateKey == nil && w.StateKey() != nil) || (g.StateKey != nil && w.StateKey() == nil) { + t.Fatalf("%s event[%d] state_key [not] missing: got %v want %v", msg, i, g.StateKey, w.StateKey()) + } + if g.StateKey != nil { + if !w.StateKeyEquals(*g.StateKey) { + t.Errorf("%s event[%d] state_key mismatch: got %s want %s", msg, i, *g.StateKey, *w.StateKey()) + } + } + } +} + +func reversed(in []gomatrixserverlib.HeaderedEvent) []gomatrixserverlib.HeaderedEvent { + out := make([]gomatrixserverlib.HeaderedEvent, len(in)) + for i := 0; i < len(in); i++ { + out[i] = in[len(in)-i-1] + } + return out +}