Only store notifications for users with pushers, de-parallelise TestSessionCleanUp for now

clientapi/routing/register_test.go

@@ -214,7 +214,7 @@ func TestSessionCleanUp(t *testing.T) {
 	s := newSessionsDict()
 
 	t.Run("session is cleaned up after a while", func(t *testing.T) {
-		t.Parallel()
+		// t.Parallel()
 		dummySession := "helloWorld"
 		// manually added, as s.addParams() would start the timer with the default timeout
 		s.params[dummySession] = registerRequest{Username: "Testing"}
@@ -226,7 +226,7 @@ func TestSessionCleanUp(t *testing.T) {
 	})
 
 	t.Run("session is deleted, once the registration completed", func(t *testing.T) {
-		t.Parallel()
+		// t.Parallel()
 		dummySession := "helloWorld2"
 		s.startTimer(time.Minute, dummySession)
 		s.deleteSession(dummySession)
@@ -236,7 +236,7 @@ func TestSessionCleanUp(t *testing.T) {
 	})
 
 	t.Run("session timer is restarted after second call", func(t *testing.T) {
-		t.Parallel()
+		// t.Parallel()
 		dummySession := "helloWorld3"
 		// the following will start a timer with the default timeout of 5min
 		s.addParams(dummySession, registerRequest{Username: "Testing"})
@@ -260,4 +260,4 @@ func TestSessionCleanUp(t *testing.T) {
 			t.Error("expected session to device to be delete")
 		}
 	})
-}
+}

userapi/consumers/syncapi_streamevent.go

@@ -139,6 +139,9 @@ func (s *OutputStreamEventConsumer) processMessage(ctx context.Context, event *g
 	// removing it means we can send all notifications to
 	// e.g. Element's Push gateway in one go.
 	for _, mem := range members {
+		if p, err := s.db.GetPushers(ctx, mem.Localpart); err != nil || len(p) == 0 {
+			continue
+		}
 		if err := s.notifyLocal(ctx, event, pos, mem, roomSize, roomName); err != nil {
 			log.WithFields(log.Fields{
 				"localpart": mem.Localpart,