Merge branch 'main' of github.com:matrix-org/dendrite into s7evink/syncresponse

dendrite-sample.monolith.yaml

@@ -37,7 +37,7 @@ global:
   # you must configure the "database" block for each component instead.
   database:
     connection_string: postgresql://username:password@hostname/dendrite?sslmode=disable
-    max_open_conns: 100
+    max_open_conns: 90
     max_idle_conns: 5
     conn_max_lifetime: -1
 

roomserver/internal/perform/perform_join.go

@@ -237,7 +237,7 @@ func (r *Joiner) performJoinRoomByID(
 	// Force a federated join if we're dealing with a pending invite
 	// and we aren't in the room.
 	isInvitePending, inviteSender, _, err := helpers.IsInvitePending(ctx, r.DB, req.RoomIDOrAlias, req.UserID)
-	if err == nil && isInvitePending {
+	if err == nil && !serverInRoom && isInvitePending {
 		_, inviterDomain, ierr := gomatrixserverlib.SplitID('@', inviteSender)
 		if ierr != nil {
 			return "", "", fmt.Errorf("gomatrixserverlib.SplitID: %w", err)

setup/base/base.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/tls"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net"
@@ -467,8 +468,13 @@ func (b *BaseDendrite) SetupAndServeHTTP(
 		w.WriteHeader(200)
 	})
 	b.DendriteAdminMux.HandleFunc("/monitor/health", func(w http.ResponseWriter, r *http.Request) {
-		if b.ProcessContext.IsDegraded() {
+		if isDegraded, reasons := b.ProcessContext.IsDegraded(); isDegraded {
 			w.WriteHeader(503)
+			_ = json.NewEncoder(w).Encode(struct {
+				Warnings []string `json:"warnings"`
+			}{
+				Warnings: reasons,
+			})
 			return
 		}
 		w.WriteHeader(200)

setup/jetstream/nats.go

@@ -169,9 +169,9 @@ func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsc
 					// We've managed to add the stream in memory.  What's on the
 					// disk will be left alone, but our ability to recover from a
 					// future crash will be limited. Yell about it.
-					sentry.CaptureException(fmt.Errorf("Stream %q is running in-memory; this may be due to data corruption in the JetStream storage directory, investigate as soon as possible", namespaced.Name))
+					err := fmt.Errorf("Stream %q is running in-memory; this may be due to data corruption in the JetStream storage directory", namespaced.Name)
-					logrus.Warn("Stream is running in-memory; this may be due to data corruption in the JetStream storage directory, investigate as soon as possible")
+					sentry.CaptureException(err)
-					process.Degraded()
+					process.Degraded(err)
 				}
 			}
 		}

setup/process/process.go

@@ -2,19 +2,18 @@ package process
 
 import (
 	"context"
-	"fmt"
 	"sync"
 
 	"github.com/getsentry/sentry-go"
 	"github.com/sirupsen/logrus"
-	"go.uber.org/atomic"
 )
 
 type ProcessContext struct {
-	wg       *sync.WaitGroup    // used to wait for components to shutdown
+	mu       sync.RWMutex
-	ctx      context.Context    // cancelled when Stop is called
+	wg       *sync.WaitGroup     // used to wait for components to shutdown
-	shutdown context.CancelFunc // shut down Dendrite
+	ctx      context.Context     // cancelled when Stop is called
-	degraded atomic.Bool
+	shutdown context.CancelFunc  // shut down Dendrite
+	degraded map[string]struct{} // reasons why the process is degraded
 }
 
 func NewProcessContext() *ProcessContext {
@@ -50,13 +49,25 @@ func (b *ProcessContext) WaitForComponentsToFinish() {
 	b.wg.Wait()
 }
 
-func (b *ProcessContext) Degraded() {
+func (b *ProcessContext) Degraded(err error) {
-	if b.degraded.CompareAndSwap(false, true) {
+	b.mu.Lock()
-		logrus.Warn("Dendrite is running in a degraded state")
+	defer b.mu.Unlock()
-		sentry.CaptureException(fmt.Errorf("Process is running in a degraded state"))
+	if _, ok := b.degraded[err.Error()]; !ok {
+		logrus.WithError(err).Warn("Dendrite has entered a degraded state")
+		sentry.CaptureException(err)
+		b.degraded[err.Error()] = struct{}{}
 	}
 }
 
-func (b *ProcessContext) IsDegraded() bool {
+func (b *ProcessContext) IsDegraded() (bool, []string) {
-	return b.degraded.Load()
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	if len(b.degraded) == 0 {
+		return false, nil
+	}
+	reasons := make([]string, 0, len(b.degraded))
+	for reason := range b.degraded {
+		reasons = append(reasons, reason)
+	}
+	return true, reasons
 }

syncapi/storage/postgres/output_room_events_table.go

@@ -76,6 +76,8 @@ CREATE INDEX IF NOT EXISTS syncapi_output_room_events_type_idx ON syncapi_output
 CREATE INDEX IF NOT EXISTS syncapi_output_room_events_sender_idx ON syncapi_output_room_events (sender);
 CREATE INDEX IF NOT EXISTS syncapi_output_room_events_room_id_idx ON syncapi_output_room_events (room_id);
 CREATE INDEX IF NOT EXISTS syncapi_output_room_events_exclude_from_sync_idx ON syncapi_output_room_events (exclude_from_sync);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_add_state_ids_idx ON syncapi_output_room_events ((add_state_ids IS NOT NULL));
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_remove_state_ids_idx ON syncapi_output_room_events ((remove_state_ids IS NOT NULL));
 `
 
 const insertEventSQL = "" +

syncapi/storage/sqlite3/output_room_events_table.go

@@ -55,6 +55,8 @@ CREATE INDEX IF NOT EXISTS syncapi_output_room_events_type_idx ON syncapi_output
 CREATE INDEX IF NOT EXISTS syncapi_output_room_events_sender_idx ON syncapi_output_room_events (sender);
 CREATE INDEX IF NOT EXISTS syncapi_output_room_events_room_id_idx ON syncapi_output_room_events (room_id);
 CREATE INDEX IF NOT EXISTS syncapi_output_room_events_exclude_from_sync_idx ON syncapi_output_room_events (exclude_from_sync);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_add_state_ids_idx ON syncapi_output_room_events ((add_state_ids IS NOT NULL));
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_remove_state_ids_idx ON syncapi_output_room_events ((remove_state_ids IS NOT NULL));
 `
 
 const insertEventSQL = "" +