Merge branch 'main' into neilalexander/retireinvite

clientapi/routing/createroom.go

@@ -49,6 +49,7 @@ type createRoomRequest struct {
 	GuestCanJoin              bool                          `json:"guest_can_join"`
 	RoomVersion               gomatrixserverlib.RoomVersion `json:"room_version"`
 	PowerLevelContentOverride json.RawMessage               `json:"power_level_content_override"`
+	IsDirect                  bool                          `json:"is_direct"`
 }
 
 const (
@@ -499,9 +500,17 @@ func createRoom(
 		// Build some stripped state for the invite.
 		var globalStrippedState []gomatrixserverlib.InviteV2StrippedState
 		for _, event := range builtEvents {
+			// Chosen events from the spec:
+			// https://spec.matrix.org/v1.3/client-server-api/#stripped-state
 			switch event.Type() {
+			case gomatrixserverlib.MRoomCreate:
+				fallthrough
 			case gomatrixserverlib.MRoomName:
 				fallthrough
+			case gomatrixserverlib.MRoomAvatar:
+				fallthrough
+			case gomatrixserverlib.MRoomTopic:
+				fallthrough
 			case gomatrixserverlib.MRoomCanonicalAlias:
 				fallthrough
 			case gomatrixserverlib.MRoomEncryption:
@@ -522,7 +531,7 @@ func createRoom(
 			// Build the invite event.
 			inviteEvent, err := buildMembershipEvent(
 				ctx, invitee, "", profileAPI, device, gomatrixserverlib.Invite,
-				roomID, true, cfg, evTime, rsAPI, asAPI,
+				roomID, r.IsDirect, cfg, evTime, rsAPI, asAPI,
 			)
 			if err != nil {
 				util.GetLogger(ctx).WithError(err).Error("buildMembershipEvent failed")

cmd/create-account/main.go

@@ -85,6 +85,10 @@ func main() {
 		logrus.Fatalf("The reset-password flag has been replaced by the POST /_dendrite/admin/resetPassword/{localpart} admin API.")
 	}
 
+	if cfg.ClientAPI.RegistrationSharedSecret == "" {
+		logrus.Fatalln("Shared secret registration is not enabled, enable it by setting a shared secret in the config: 'client_api.registration_shared_secret'")
+	}
+
 	if *username == "" {
 		flag.Usage()
 		os.Exit(1)

roomserver/internal/input/input_events.go

@@ -17,8 +17,8 @@
 package input
 
 import (
-	"bytes"
 	"context"
+	"database/sql"
 	"fmt"
 	"time"
 
@@ -107,28 +107,6 @@ func (r *Inputer) processRoomEvent(
 		})
 	}
 
-	// if we have already got this event then do not process it again, if the input kind is an outlier.
-	// Outliers contain no extra information which may warrant a re-processing.
-	if input.Kind == api.KindOutlier {
-		evs, err2 := r.DB.EventsFromIDs(ctx, []string{event.EventID()})
-		if err2 == nil && len(evs) == 1 {
-			// check hash matches if we're on early room versions where the event ID was a random string
-			idFormat, err2 := headered.RoomVersion.EventIDFormat()
-			if err2 == nil {
-				switch idFormat {
-				case gomatrixserverlib.EventIDFormatV1:
-					if bytes.Equal(event.EventReference().EventSHA256, evs[0].EventReference().EventSHA256) {
-						logger.Debugf("Already processed event; ignoring")
-						return nil
-					}
-				default:
-					logger.Debugf("Already processed event; ignoring")
-					return nil
-				}
-			}
-		}
-	}
-
 	// Don't waste time processing the event if the room doesn't exist.
 	// A room entry locally will only be created in response to a create
 	// event.
@@ -141,6 +119,29 @@ func (r *Inputer) processRoomEvent(
 		return fmt.Errorf("room %s does not exist for event %s", event.RoomID(), event.EventID())
 	}
 
+	// If we already know about this outlier and it hasn't been rejected
+	// then we won't attempt to reprocess it. If it was rejected or has now
+	// arrived as a different kind of event, then we can attempt to reprocess,
+	// in case we have learned something new or need to weave the event into
+	// the DAG now.
+	if input.Kind == api.KindOutlier && roomInfo != nil {
+		wasRejected, werr := r.DB.IsEventRejected(ctx, roomInfo.RoomNID, event.EventID())
+		switch {
+		case werr == sql.ErrNoRows:
+			// We haven't seen this event before so continue.
+		case werr != nil:
+			// Something has gone wrong trying to find out if we rejected
+			// this event already.
+			logger.WithError(werr).Errorf("Failed to check if event %q is already seen", event.EventID())
+			return werr
+		case !wasRejected:
+			// We've seen this event before and it wasn't rejected so we
+			// should ignore it.
+			logger.Debugf("Already processed event %q, ignoring", event.EventID())
+			return nil
+		}
+	}
+
 	var missingAuth, missingPrev bool
 	serverRes := &fedapi.QueryJoinedHostServerNamesInRoomResponse{}
 	if !isCreateEvent {

roomserver/internal/query/query.go

@@ -72,13 +72,10 @@ func (r *Queryer) QueryStateAfterEvents(
 
 	prevStates, err := r.DB.StateAtEventIDs(ctx, request.PrevEventIDs)
 	if err != nil {
-		switch err.(type) {
+		if _, ok := err.(types.MissingEventError); ok {
-		case types.MissingEventError:
-			util.GetLogger(ctx).Errorf("QueryStateAfterEvents: MissingEventError: %s", err)
 			return nil
-		default:
-			return err
 		}
+		return err
 	}
 	response.PrevEventsExist = true
 
@@ -95,6 +92,12 @@ func (r *Queryer) QueryStateAfterEvents(
 		)
 	}
 	if err != nil {
+		if _, ok := err.(types.MissingEventError); ok {
+			return nil
+		}
+		if _, ok := err.(types.MissingStateError); ok {
+			return nil
+		}
 		return err
 	}
 

roomserver/storage/interface.go

@@ -94,6 +94,8 @@ type Database interface {
 	// Opens and returns a room updater, which locks the room and opens a transaction.
 	// The GetRoomUpdater must have Commit or Rollback called on it if this doesn't return an error.
 	// If this returns an error then no further action is required.
+	// IsEventRejected returns true if the event is known and rejected.
+	IsEventRejected(ctx context.Context, roomNID types.RoomNID, eventID string) (rejected bool, err error)
 	GetRoomUpdater(ctx context.Context, roomInfo *types.RoomInfo) (*shared.RoomUpdater, error)
 	// Look up event references for the latest events in the room and the current state snapshot.
 	// Returns the latest events, the current state and the maximum depth of the latest events plus 1.

roomserver/storage/postgres/events_table.go

@@ -136,6 +136,9 @@ const selectMaxEventDepthSQL = "" +
 const selectRoomNIDsForEventNIDsSQL = "" +
 	"SELECT event_nid, room_nid FROM roomserver_events WHERE event_nid = ANY($1)"
 
+const selectEventRejectedSQL = "" +
+	"SELECT is_rejected FROM roomserver_events WHERE room_nid = $1 AND event_id = $2"
+
 type eventStatements struct {
 	insertEventStmt                        *sql.Stmt
 	selectEventStmt                        *sql.Stmt
@@ -153,6 +156,7 @@ type eventStatements struct {
 	bulkSelectUnsentEventNIDStmt           *sql.Stmt
 	selectMaxEventDepthStmt                *sql.Stmt
 	selectRoomNIDsForEventNIDsStmt         *sql.Stmt
+	selectEventRejectedStmt                *sql.Stmt
 }
 
 func CreateEventsTable(db *sql.DB) error {
@@ -180,6 +184,7 @@ func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
 		{&s.bulkSelectUnsentEventNIDStmt, bulkSelectUnsentEventNIDSQL},
 		{&s.selectMaxEventDepthStmt, selectMaxEventDepthSQL},
 		{&s.selectRoomNIDsForEventNIDsStmt, selectRoomNIDsForEventNIDsSQL},
+		{&s.selectEventRejectedStmt, selectEventRejectedSQL},
 	}.Prepare(db)
 }
 
@@ -540,3 +545,11 @@ func eventNIDsAsArray(eventNIDs []types.EventNID) pq.Int64Array {
 	}
 	return nids
 }
+
+func (s *eventStatements) SelectEventRejected(
+	ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventID string,
+) (rejected bool, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectEventRejectedStmt)
+	err = stmt.QueryRowContext(ctx, roomNID, eventID).Scan(&rejected)
+	return
+}

roomserver/storage/shared/storage.go

@@ -567,6 +567,10 @@ func (d *Database) GetRoomUpdater(
 	return updater, err
 }
 
+func (d *Database) IsEventRejected(ctx context.Context, roomNID types.RoomNID, eventID string) (bool, error) {
+	return d.EventsTable.SelectEventRejected(ctx, nil, roomNID, eventID)
+}
+
 func (d *Database) StoreEvent(
 	ctx context.Context, event *gomatrixserverlib.Event,
 	authEventNIDs []types.EventNID, isRejected bool,

roomserver/storage/sqlite3/events_table.go

@@ -109,6 +109,9 @@ const selectMaxEventDepthSQL = "" +
 const selectRoomNIDsForEventNIDsSQL = "" +
 	"SELECT event_nid, room_nid FROM roomserver_events WHERE event_nid IN ($1)"
 
+const selectEventRejectedSQL = "" +
+	"SELECT is_rejected FROM roomserver_events WHERE room_nid = $1 AND event_id = $2"
+
 type eventStatements struct {
 	db                                     *sql.DB
 	insertEventStmt                        *sql.Stmt
@@ -122,6 +125,7 @@ type eventStatements struct {
 	bulkSelectStateAtEventAndReferenceStmt *sql.Stmt
 	bulkSelectEventReferenceStmt           *sql.Stmt
 	bulkSelectEventIDStmt                  *sql.Stmt
+	selectEventRejectedStmt                *sql.Stmt
 	//bulkSelectEventNIDStmt               *sql.Stmt
 	//bulkSelectUnsentEventNIDStmt         *sql.Stmt
 	//selectRoomNIDsForEventNIDsStmt       *sql.Stmt
@@ -152,6 +156,7 @@ func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
 		//{&s.bulkSelectEventNIDStmt, bulkSelectEventNIDSQL},
 		//{&s.bulkSelectUnsentEventNIDStmt, bulkSelectUnsentEventNIDSQL},
 		//{&s.selectRoomNIDForEventNIDStmt, selectRoomNIDForEventNIDSQL},
+		{&s.selectEventRejectedStmt, selectEventRejectedSQL},
 	}.Prepare(db)
 }
 
@@ -614,3 +619,11 @@ func eventNIDsAsArray(eventNIDs []types.EventNID) string {
 	b, _ := json.Marshal(eventNIDs)
 	return string(b)
 }
+
+func (s *eventStatements) SelectEventRejected(
+	ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventID string,
+) (rejected bool, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectEventRejectedStmt)
+	err = stmt.QueryRowContext(ctx, roomNID, eventID).Scan(&rejected)
+	return
+}

roomserver/storage/tables/interface.go

@@ -66,6 +66,7 @@ type Events interface {
 	BulkSelectUnsentEventNID(ctx context.Context, txn *sql.Tx, eventIDs []string) (map[string]types.EventNID, error)
 	SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error)
 	SelectRoomNIDsForEventNIDs(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (roomNIDs map[types.EventNID]types.RoomNID, err error)
+	SelectEventRejected(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventID string) (rejected bool, err error)
 }
 
 type Rooms interface {

syncapi/storage/postgres/output_room_events_table.go

@@ -279,8 +279,8 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
 			log.WithFields(log.Fields{
 				"since":   r.From,
 				"current": r.To,
-				"adds":    addIDs,
+				"adds":    len(addIDs),
-				"dels":    delIDs,
+				"dels":    len(delIDs),
 			}).Warn("StateBetween: ignoring deleted state")
 		}
 

syncapi/storage/sqlite3/output_room_events_table.go

@@ -234,8 +234,8 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
 			log.WithFields(log.Fields{
 				"since":   r.From,
 				"current": r.To,
-				"adds":    addIDsJSON,
+				"adds":    len(addIDsJSON),
-				"dels":    delIDsJSON,
+				"dels":    len(delIDsJSON),
 			}).Warn("StateBetween: ignoring deleted state")
 		}
 

syncapi/streams/stream_pdu.go

@@ -560,14 +560,13 @@ func (p *PDUStreamProvider) lazyLoadMembers(
 			// If this is a gapped incremental sync, we still want this membership
 			isGappedIncremental := limited && incremental
 			// We want this users membership event, keep it in the list
-			_, ok := timelineUsers[event.Sender()]
+			stateKey := *event.StateKey()
-			wantMembership := ok || isGappedIncremental
+			if _, ok := timelineUsers[stateKey]; ok || isGappedIncremental {
-			if wantMembership {
 				newStateEvents = append(newStateEvents, event)
 				if !includeRedundant {
-					p.lazyLoadCache.StoreLazyLoadedUser(device, roomID, event.Sender(), event.EventID())
+					p.lazyLoadCache.StoreLazyLoadedUser(device, roomID, stateKey, event.EventID())
 				}
-				delete(timelineUsers, event.Sender())
+				delete(timelineUsers, stateKey)
 			}
 		} else {
 			newStateEvents = append(newStateEvents, event)
@@ -578,17 +577,16 @@ func (p *PDUStreamProvider) lazyLoadMembers(
 		wantUsers = append(wantUsers, userID)
 	}
 	// Query missing membership events
-	memberships, err := p.DB.GetStateEventsForRoom(ctx, roomID, &gomatrixserverlib.StateFilter{
+	filter := gomatrixserverlib.DefaultStateFilter()
-		Limit:   100,
+	filter.Senders = &wantUsers
-		Senders: &wantUsers,
+	filter.Types = &[]string{gomatrixserverlib.MRoomMember}
-		Types:   &[]string{gomatrixserverlib.MRoomMember},
+	memberships, err := p.DB.GetStateEventsForRoom(ctx, roomID, &filter)
-	})
 	if err != nil {
 		return stateEvents, err
 	}
 	// cache the membership events
 	for _, membership := range memberships {
-		p.lazyLoadCache.StoreLazyLoadedUser(device, roomID, membership.Sender(), membership.EventID())
+		p.lazyLoadCache.StoreLazyLoadedUser(device, roomID, *membership.StateKey(), membership.EventID())
 	}
 	stateEvents = append(newStateEvents, memberships...)
 	return stateEvents, nil

syncapi/syncapi_test.go

@@ -10,6 +10,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/nats-io/nats.go"
+	"github.com/tidwall/gjson"
+
 	"github.com/matrix-org/dendrite/clientapi/producers"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/roomserver"
@@ -21,9 +25,6 @@ import (
 	"github.com/matrix-org/dendrite/test"
 	"github.com/matrix-org/dendrite/test/testrig"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/nats-io/nats.go"
-	"github.com/tidwall/gjson"
 )
 
 type syncRoomserverAPI struct {
@@ -422,6 +423,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
 				if err := api.SendEvents(ctx, rsAPI, api.KindNew, eventsToSend, "test", "test", nil, false); err != nil {
 					t.Fatalf("failed to send events: %v", err)
 				}
+				time.Sleep(100 * time.Millisecond) // TODO: find a better way
 
 				// There is only one event, we expect only to be able to see this, if the room is world_readable
 				w := httptest.NewRecorder()
@@ -454,6 +456,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
 				if err := api.SendEvents(ctx, rsAPI, api.KindNew, eventsToSend, "test", "test", nil, false); err != nil {
 					t.Fatalf("failed to send events: %v", err)
 				}
+				time.Sleep(100 * time.Millisecond) // TODO: find a better way
 
 				// Verify the messages after/before invite are visible or not
 				w = httptest.NewRecorder()