mirror of
https://github.com/matrix-org/dendrite.git
synced 2026-01-11 16:13:10 -06:00
Merge branch 'main' into neilalexander/cfg
This commit is contained in:
Neil Alexander
GitHub
commit
e2d97d4ab9
30
CHANGES.md
30
CHANGES.md
|
|
@ -1,5 +1,35 @@
|
|||
# Changelog
|
||||
|
||||
## Dendrite 0.9.0 (2022-08-01)
|
||||
|
||||
### Features
|
||||
|
||||
* Dendrite now uses Ristretto for managing in-memory caches
|
||||
* Should improve cache utilisation considerably over time by more intelligently selecting and managing cache entries compared to the previous LRU-based cache
|
||||
* Defaults to a 1GB cache size if not configured otherwise
|
||||
* The estimated cache size in memory and maximum age can now be configured with new [configuration options](https://github.com/matrix-org/dendrite/blob/e94ef84aaba30e12baf7f524c4e7a36d2fdeb189/dendrite-sample.monolith.yaml#L44-L61) to prevent unbounded cache growth
|
||||
* Added support for serving the `/.well-known/matrix/client` hint directly from Dendrite
|
||||
* Configurable with the new [configuration option](https://github.com/matrix-org/dendrite/blob/e94ef84aaba30e12baf7f524c4e7a36d2fdeb189/dendrite-sample.monolith.yaml#L67-L69)
|
||||
* Refactored membership updater, which should eliminate some bugs caused by the membership table getting out of sync with the room state
|
||||
* The User API is now responsible for sending account data updates to other components, which may fix some races and duplicate account data events
|
||||
* Optimised database query for checking whether a remote server is allowed to request an event over federation without using anywhere near as much CPU time (PostgreSQL only)
|
||||
* Database migrations have been refactored to eliminate some problems that were present with `goose` and upgrading from older Dendrite versions
|
||||
* Media fetching will now use the `/v3` endpoints for downloading media from remote homeservers
|
||||
* HTTP 404 and HTTP 405 errors from the client-facing APIs should now be returned with CORS headers so that web-based clients do not produce incorrect access control warnings for unknown endpoints
|
||||
* Some preparation work for full history visibility support
|
||||
|
||||
### Fixes
|
||||
|
||||
* Fixes a crash that could occur during event redaction
|
||||
* The `/members` endpoint will no longer incorrectly return HTTP 500 as a result of some invite events
|
||||
* Send-to-device messages should now be ordered more reliably and the last position in the stream updated correctly
|
||||
* Parsing of appservice configuration files is now less strict (contributed by [Kab1r](https://github.com/Kab1r))
|
||||
* The sync API should now identify shared users correctly when waking up for E2EE key changes
|
||||
* The federation `/state` endpoint will now return a HTTP 403 when the state before an event isn't known instead of a HTTP 500
|
||||
* Presence timestamps should now be calculated with the correct precision
|
||||
* A race condition in the roomserver's room info has been fixed
|
||||
* A race condition in the sync API has been fixed
|
||||
|
||||
## Dendrite 0.8.9 (2022-07-01)
|
||||
|
||||
### Features
|
||||
|
|
|
|||
|
|
@ -48,7 +48,6 @@ func AddPublicRoutes(
|
|||
|
||||
syncProducer := &producers.SyncAPIProducer{
|
||||
JetStream: js,
|
||||
TopicClientData: cfg.Matrix.JetStream.Prefixed(jetstream.OutputClientData),
|
||||
TopicReceiptEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
|
||||
TopicSendToDeviceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
|
||||
TopicTypingEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ import (
|
|||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal/eventutil"
|
||||
"github.com/matrix-org/dendrite/setup/jetstream"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
userapi "github.com/matrix-org/dendrite/userapi/api"
|
||||
|
|
@ -32,7 +31,6 @@ import (
|
|||
|
||||
// SyncAPIProducer produces events for the sync API server to consume
|
||||
type SyncAPIProducer struct {
|
||||
TopicClientData string
|
||||
TopicReceiptEvent string
|
||||
TopicSendToDeviceEvent string
|
||||
TopicTypingEvent string
|
||||
|
|
@ -42,36 +40,6 @@ type SyncAPIProducer struct {
|
|||
UserAPI userapi.ClientUserAPI
|
||||
}
|
||||
|
||||
// SendData sends account data to the sync API server
|
||||
func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string, readMarker *eventutil.ReadMarkerJSON, ignoredUsers *types.IgnoredUsers) error {
|
||||
m := &nats.Msg{
|
||||
Subject: p.TopicClientData,
|
||||
Header: nats.Header{},
|
||||
}
|
||||
m.Header.Set(jetstream.UserID, userID)
|
||||
|
||||
data := eventutil.AccountData{
|
||||
RoomID: roomID,
|
||||
Type: dataType,
|
||||
ReadMarker: readMarker,
|
||||
IgnoredUsers: ignoredUsers,
|
||||
}
|
||||
var err error
|
||||
m.Data, err = json.Marshal(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"user_id": userID,
|
||||
"room_id": roomID,
|
||||
"data_type": dataType,
|
||||
}).Tracef("Producing to topic '%s'", p.TopicClientData)
|
||||
|
||||
_, err = p.JetStream.PublishMsg(m)
|
||||
return err
|
||||
}
|
||||
|
||||
func (p *SyncAPIProducer) SendReceipt(
|
||||
ctx context.Context,
|
||||
userID, roomID, eventID, receiptType string, timestamp gomatrixserverlib.Timestamp,
|
||||
|
|
|
|||
|
|
@ -25,7 +25,6 @@ import (
|
|||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/internal/eventutil"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/dendrite/userapi/api"
|
||||
|
||||
"github.com/matrix-org/util"
|
||||
|
|
@ -127,18 +126,6 @@ func SaveAccountData(
|
|||
return util.ErrorResponse(err)
|
||||
}
|
||||
|
||||
var ignoredUsers *types.IgnoredUsers
|
||||
if dataType == "m.ignored_user_list" {
|
||||
ignoredUsers = &types.IgnoredUsers{}
|
||||
_ = json.Unmarshal(body, ignoredUsers)
|
||||
}
|
||||
|
||||
// TODO: user API should do this since it's account data
|
||||
if err := syncProducer.SendData(userID, roomID, dataType, nil, ignoredUsers); err != nil {
|
||||
util.GetLogger(req.Context()).WithError(err).Error("syncProducer.SendData failed")
|
||||
return jsonerror.InternalServerError()
|
||||
}
|
||||
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusOK,
|
||||
JSON: struct{}{},
|
||||
|
|
@ -191,11 +178,6 @@ func SaveReadMarker(
|
|||
return util.ErrorResponse(err)
|
||||
}
|
||||
|
||||
if err := syncProducer.SendData(device.UserID, roomID, "m.fully_read", &r, nil); err != nil {
|
||||
util.GetLogger(req.Context()).WithError(err).Error("syncProducer.SendData failed")
|
||||
return jsonerror.InternalServerError()
|
||||
}
|
||||
|
||||
// Handle the read receipt that may be included in the read marker
|
||||
if r.Read != "" {
|
||||
return SetReceipt(req, syncProducer, device, roomID, "m.read", r.Read)
|
||||
|
|
|
|||
|
|
@ -18,8 +18,6 @@ import (
|
|||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/httputil"
|
||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
|
|
@ -98,10 +96,6 @@ func PutTag(
|
|||
return jsonerror.InternalServerError()
|
||||
}
|
||||
|
||||
if err = syncProducer.SendData(userID, roomID, "m.tag", nil, nil); err != nil {
|
||||
logrus.WithError(err).Error("Failed to send m.tag account data update to syncapi")
|
||||
}
|
||||
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusOK,
|
||||
JSON: struct{}{},
|
||||
|
|
@ -150,11 +144,6 @@ func DeleteTag(
|
|||
return jsonerror.InternalServerError()
|
||||
}
|
||||
|
||||
// TODO: user API should do this since it's account data
|
||||
if err := syncProducer.SendData(userID, roomID, "m.tag", nil, nil); err != nil {
|
||||
logrus.WithError(err).Error("Failed to send m.tag account data update to syncapi")
|
||||
}
|
||||
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusOK,
|
||||
JSON: struct{}{},
|
||||
|
|
|
|||
14
go.mod
14
go.mod
|
|
@ -1,9 +1,5 @@
|
|||
module github.com/matrix-org/dendrite
|
||||
|
||||
replace github.com/nats-io/nats-server/v2 => github.com/neilalexander/nats-server/v2 v2.8.3-0.20220513095553-73a9a246d34f
|
||||
|
||||
replace github.com/nats-io/nats.go => github.com/neilalexander/nats.go v1.13.1-0.20220621084451-ac518c356673
|
||||
|
||||
require (
|
||||
github.com/Arceliar/ironwood v0.0.0-20220306165321-319147a02d98
|
||||
github.com/Arceliar/phony v0.0.0-20210209235338-dde1a8dca979
|
||||
|
|
@ -29,8 +25,8 @@ require (
|
|||
github.com/matrix-org/pinecone v0.0.0-20220708135211-1ce778fcde6a
|
||||
github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
|
||||
github.com/mattn/go-sqlite3 v1.14.13
|
||||
github.com/nats-io/nats-server/v2 v2.7.4-0.20220309205833-773636c1c5bb
|
||||
github.com/nats-io/nats.go v1.14.0
|
||||
github.com/nats-io/nats-server/v2 v2.8.5-0.20220731184415-903a06a5b4ee
|
||||
github.com/nats-io/nats.go v1.16.1-0.20220731182438-87bbea85922b
|
||||
github.com/neilalexander/utp v0.1.1-0.20210727203401-54ae7b1cd5f9
|
||||
github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
|
||||
github.com/ngrok/sqlmw v0.0.0-20220520173518-97c9c04efc79
|
||||
|
|
@ -76,7 +72,7 @@ require (
|
|||
github.com/h2non/filetype v1.1.3 // indirect
|
||||
github.com/juju/errors v0.0.0-20220203013757-bd733f3c86b9 // indirect
|
||||
github.com/juju/testing v0.0.0-20220203020004-a0ff61f03494 // indirect
|
||||
github.com/klauspost/compress v1.14.4 // indirect
|
||||
github.com/klauspost/compress v1.15.9 // indirect
|
||||
github.com/lucas-clemente/quic-go v0.26.0 // indirect
|
||||
github.com/marten-seemann/qtls-go1-16 v0.1.5 // indirect
|
||||
github.com/marten-seemann/qtls-go1-17 v0.1.1 // indirect
|
||||
|
|
@ -86,7 +82,7 @@ require (
|
|||
github.com/minio/highwayhash v1.0.2 // indirect
|
||||
github.com/moby/term v0.0.0-20210610120745-9d4ed1856297 // indirect
|
||||
github.com/morikuni/aec v1.0.0 // indirect
|
||||
github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a // indirect
|
||||
github.com/nats-io/jwt/v2 v2.3.0 // indirect
|
||||
github.com/nats-io/nkeys v0.3.0 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
github.com/nxadm/tail v1.4.8 // indirect
|
||||
|
|
@ -104,7 +100,7 @@ require (
|
|||
golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
|
||||
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
|
||||
golang.org/x/text v0.3.8-0.20211004125949-5bd84dd9b33b // indirect
|
||||
golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect
|
||||
golang.org/x/time v0.0.0-20220411224347-583f2d630306 // indirect
|
||||
golang.org/x/tools v0.1.10 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
|
||||
google.golang.org/protobuf v1.27.1 // indirect
|
||||
|
|
|
|||
20
go.sum
20
go.sum
|
|
@ -302,8 +302,8 @@ github.com/kardianos/minwinsvc v1.0.0/go.mod h1:Bgd0oc+D0Qo3bBytmNtyRKVlp85dAloL
|
|||
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/klauspost/compress v1.10.3/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
|
||||
github.com/klauspost/compress v1.14.4 h1:eijASRJcobkVtSt81Olfh7JX43osYLwy5krOJo6YEu4=
|
||||
github.com/klauspost/compress v1.14.4/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
|
||||
github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
|
||||
github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
|
|
@ -381,8 +381,12 @@ github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7P
|
|||
github.com/mschoch/smat v0.0.0-20160514031455-90eadee771ae/go.mod h1:qAyveg+e4CE+eKJXWVjKXM4ck2QobLqTDytGJbLLhJg=
|
||||
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
|
||||
github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
|
||||
github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a h1:lem6QCvxR0Y28gth9P+wV2K/zYUUAkJ+55U8cpS0p5I=
|
||||
github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
|
||||
github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
|
||||
github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
|
||||
github.com/nats-io/nats-server/v2 v2.8.5-0.20220731184415-903a06a5b4ee h1:vAtoZ+LW6eIUjkCWWwO1DZ6o16UGrVOG+ot/AkwejO8=
|
||||
github.com/nats-io/nats-server/v2 v2.8.5-0.20220731184415-903a06a5b4ee/go.mod h1:3Yg3ApyQxPlAs1KKHKV5pobV5VtZk+TtOiUJx/iqkkg=
|
||||
github.com/nats-io/nats.go v1.16.1-0.20220731182438-87bbea85922b h1:CE9wSYLvwq8aC/0+6zH8lhhtZYvJ9p8PzwvZeYgdBc0=
|
||||
github.com/nats-io/nats.go v1.16.1-0.20220731182438-87bbea85922b/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
|
||||
github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
|
||||
github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
|
|
@ -390,10 +394,6 @@ github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OS
|
|||
github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32/go.mod h1:9wM+0iRr9ahx58uYLpLIr5fm8diHn0JbqRycJi6w0Ms=
|
||||
github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJEU3ofeGjhHklVoIGuVj85JJwZ6kWPaJwCIxgnFmo=
|
||||
github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM=
|
||||
github.com/neilalexander/nats-server/v2 v2.8.3-0.20220513095553-73a9a246d34f h1:Fc+TjdV1mOy0oISSzfoxNWdTqjg7tN/Vdgf+B2cwvdo=
|
||||
github.com/neilalexander/nats-server/v2 v2.8.3-0.20220513095553-73a9a246d34f/go.mod h1:vIdpKz3OG+DCg4q/xVPdXHoztEyKDWRtykQ4N7hd7C4=
|
||||
github.com/neilalexander/nats.go v1.13.1-0.20220621084451-ac518c356673 h1:TcKfa3Tf0qwUotv63PQVu2d1bBoLi2iEA4RHVMGDh5M=
|
||||
github.com/neilalexander/nats.go v1.13.1-0.20220621084451-ac518c356673/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
|
||||
github.com/neilalexander/utp v0.1.1-0.20210727203401-54ae7b1cd5f9 h1:lrVQzBtkeQEGGYUHwSX1XPe1E5GL6U3KYCNe2G4bncQ=
|
||||
github.com/neilalexander/utp v0.1.1-0.20210727203401-54ae7b1cd5f9/go.mod h1:NPHGhPc0/wudcaCqL/H5AOddkRf8GPRhzOujuUKGQu8=
|
||||
github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
|
||||
|
|
@ -764,8 +764,8 @@ golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxb
|
|||
golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 h1:GZokNIeuVkl3aZHJchRrr13WCsols02MLUcz1U9is6M=
|
||||
golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20220411224347-583f2d630306 h1:+gHMid33q6pen7kv9xvT+JRinntgeXO2AeZVd0AWD3w=
|
||||
golang.org/x/time v0.0.0-20220411224347-583f2d630306/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
|
|
|
|||
|
|
@ -16,8 +16,8 @@ var build string
|
|||
|
||||
const (
|
||||
VersionMajor = 0
|
||||
VersionMinor = 8
|
||||
VersionPatch = 9
|
||||
VersionMinor = 9
|
||||
VersionPatch = 0
|
||||
VersionTag = "" // example: "rc1"
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -50,14 +50,14 @@ func CheckForSoftFail(
|
|||
if err != nil {
|
||||
return false, fmt.Errorf("db.RoomNID: %w", err)
|
||||
}
|
||||
if roomInfo == nil || roomInfo.IsStub {
|
||||
if roomInfo == nil || roomInfo.IsStub() {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Then get the state entries for the current state snapshot.
|
||||
// We'll use this to check if the event is allowed right now.
|
||||
roomState := state.NewStateResolution(db, roomInfo)
|
||||
authStateEntries, err = roomState.LoadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID)
|
||||
authStateEntries, err = roomState.LoadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID())
|
||||
if err != nil {
|
||||
return true, fmt.Errorf("roomState.LoadStateAtSnapshot: %w", err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -236,13 +236,34 @@ func LoadStateEvents(
|
|||
func CheckServerAllowedToSeeEvent(
|
||||
ctx context.Context, db storage.Database, info *types.RoomInfo, eventID string, serverName gomatrixserverlib.ServerName, isServerInRoom bool,
|
||||
) (bool, error) {
|
||||
stateAtEvent, err := db.GetHistoryVisibilityState(ctx, info, eventID, string(serverName))
|
||||
switch err {
|
||||
case nil:
|
||||
// No error, so continue normally
|
||||
case tables.OptimisationNotSupportedError:
|
||||
// The database engine didn't support this optimisation, so fall back to using
|
||||
// the old and slow method
|
||||
stateAtEvent, err = slowGetHistoryVisibilityState(ctx, db, info, eventID, serverName)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
default:
|
||||
// Something else went wrong
|
||||
return false, err
|
||||
}
|
||||
return auth.IsServerAllowed(serverName, isServerInRoom, stateAtEvent), nil
|
||||
}
|
||||
|
||||
func slowGetHistoryVisibilityState(
|
||||
ctx context.Context, db storage.Database, info *types.RoomInfo, eventID string, serverName gomatrixserverlib.ServerName,
|
||||
) ([]*gomatrixserverlib.Event, error) {
|
||||
roomState := state.NewStateResolution(db, info)
|
||||
stateEntries, err := roomState.LoadStateAtEvent(ctx, eventID)
|
||||
if err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return false, nil
|
||||
return nil, nil
|
||||
}
|
||||
return false, fmt.Errorf("roomState.LoadStateAtEvent: %w", err)
|
||||
return nil, fmt.Errorf("roomState.LoadStateAtEvent: %w", err)
|
||||
}
|
||||
|
||||
// Extract all of the event state key NIDs from the room state.
|
||||
|
|
@ -254,7 +275,7 @@ func CheckServerAllowedToSeeEvent(
|
|||
// Then request those state key NIDs from the database.
|
||||
stateKeys, err := db.EventStateKeys(ctx, stateKeyNIDs)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("db.EventStateKeys: %w", err)
|
||||
return nil, fmt.Errorf("db.EventStateKeys: %w", err)
|
||||
}
|
||||
|
||||
// If the event state key doesn't match the given servername
|
||||
|
|
@ -277,15 +298,10 @@ func CheckServerAllowedToSeeEvent(
|
|||
}
|
||||
|
||||
if len(filteredEntries) == 0 {
|
||||
return false, nil
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
stateAtEvent, err := LoadStateEvents(ctx, db, filteredEntries)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
return auth.IsServerAllowed(serverName, isServerInRoom, stateAtEvent), nil
|
||||
return LoadStateEvents(ctx, db, filteredEntries)
|
||||
}
|
||||
|
||||
// TODO: Remove this when we have tests to assert correctness of this function
|
||||
|
|
@ -393,7 +409,7 @@ func QueryLatestEventsAndState(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if roomInfo == nil || roomInfo.IsStub {
|
||||
if roomInfo == nil || roomInfo.IsStub() {
|
||||
response.RoomExists = false
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -52,7 +52,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
|
|||
}
|
||||
return
|
||||
}
|
||||
if roomInfo == nil || roomInfo.IsStub {
|
||||
if roomInfo == nil || roomInfo.IsStub() {
|
||||
res.Error = &api.PerformError{
|
||||
Code: api.PerformErrorNoRoom,
|
||||
Msg: fmt.Sprintf("Room %s not found", req.RoomID),
|
||||
|
|
|
|||
|
|
@ -73,7 +73,7 @@ func (r *Backfiller) PerformBackfill(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return fmt.Errorf("PerformBackfill: missing room info for room %s", request.RoomID)
|
||||
}
|
||||
|
||||
|
|
@ -106,7 +106,7 @@ func (r *Backfiller) backfillViaFederation(ctx context.Context, req *api.Perform
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return fmt.Errorf("backfillViaFederation: missing room info for room %s", req.RoomID)
|
||||
}
|
||||
requester := newBackfillRequester(r.DB, r.FSAPI, r.ServerName, req.BackwardsExtremities, r.PreferServers)
|
||||
|
|
@ -434,7 +434,7 @@ FindSuccessor:
|
|||
logrus.WithError(err).WithField("room_id", roomID).Error("ServersAtEvent: failed to get RoomInfo for room")
|
||||
return nil
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
logrus.WithField("room_id", roomID).Error("ServersAtEvent: failed to get RoomInfo for room, room is missing")
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ func (r *InboundPeeker) PerformInboundPeek(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return nil
|
||||
}
|
||||
response.RoomExists = true
|
||||
|
|
|
|||
|
|
@ -125,7 +125,7 @@ func (r *Inviter) PerformInvite(
|
|||
return outputUpdates, nil
|
||||
}
|
||||
|
||||
if (info == nil || info.IsStub) && !isOriginLocal && isTargetLocal {
|
||||
if (info == nil || info.IsStub()) && !isOriginLocal && isTargetLocal {
|
||||
// The invite came in over federation for a room that we don't know about
|
||||
// yet. We need to handle this a bit differently to most invites because
|
||||
// we don't know the room state, therefore the roomserver can't process
|
||||
|
|
@ -276,7 +276,7 @@ func buildInviteStrippedState(
|
|||
}
|
||||
roomState := state.NewStateResolution(db, info)
|
||||
stateEntries, err := roomState.LoadStateAtSnapshotForStringTuples(
|
||||
ctx, info.StateSnapshotNID, stateWanted,
|
||||
ctx, info.StateSnapshotNID(), stateWanted,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
|
|||
|
|
@ -61,7 +61,7 @@ func (r *Queryer) QueryStateAfterEvents(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -302,7 +302,7 @@ func (r *Queryer) QueryServerJoinedToRoom(
|
|||
if err != nil {
|
||||
return fmt.Errorf("r.DB.RoomInfo: %w", err)
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return nil
|
||||
}
|
||||
response.RoomExists = true
|
||||
|
|
@ -390,7 +390,7 @@ func (r *Queryer) QueryMissingEvents(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return fmt.Errorf("missing RoomInfo for room %s", events[0].RoomID())
|
||||
}
|
||||
|
||||
|
|
@ -429,7 +429,7 @@ func (r *Queryer) QueryStateAndAuthChain(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil || info.IsStub {
|
||||
if info == nil || info.IsStub() {
|
||||
return nil
|
||||
}
|
||||
response.RoomExists = true
|
||||
|
|
@ -774,7 +774,7 @@ func (r *Queryer) QueryRestrictedJoinAllowed(ctx context.Context, req *api.Query
|
|||
if err != nil {
|
||||
return fmt.Errorf("r.DB.RoomInfo: %w", err)
|
||||
}
|
||||
if roomInfo == nil || roomInfo.IsStub {
|
||||
if roomInfo == nil || roomInfo.IsStub() {
|
||||
return nil // fmt.Errorf("room %q doesn't exist or is stub room", req.RoomID)
|
||||
}
|
||||
// If the room version doesn't allow restricted joins then don't
|
||||
|
|
@ -837,7 +837,7 @@ func (r *Queryer) QueryRestrictedJoinAllowed(ctx context.Context, req *api.Query
|
|||
// See if the room exists. If it doesn't exist or if it's a stub
|
||||
// room entry then we can't check memberships.
|
||||
targetRoomInfo, err := r.DB.RoomInfo(ctx, rule.RoomID)
|
||||
if err != nil || targetRoomInfo == nil || targetRoomInfo.IsStub {
|
||||
if err != nil || targetRoomInfo == nil || targetRoomInfo.IsStub() {
|
||||
res.Resident = false
|
||||
continue
|
||||
}
|
||||
|
|
|
|||
|
|
@ -124,6 +124,29 @@ func (v *StateResolution) LoadStateAtEvent(
|
|||
return stateEntries, nil
|
||||
}
|
||||
|
||||
// LoadStateAtEvent loads the full state of a room before a particular event.
|
||||
func (v *StateResolution) LoadStateAtEventForHistoryVisibility(
|
||||
ctx context.Context, eventID string,
|
||||
) ([]types.StateEntry, error) {
|
||||
span, ctx := opentracing.StartSpanFromContext(ctx, "StateResolution.LoadStateAtEvent")
|
||||
defer span.Finish()
|
||||
|
||||
snapshotNID, err := v.db.SnapshotNIDFromEventID(ctx, eventID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("LoadStateAtEvent.SnapshotNIDFromEventID failed for event %s : %w", eventID, err)
|
||||
}
|
||||
if snapshotNID == 0 {
|
||||
return nil, fmt.Errorf("LoadStateAtEvent.SnapshotNIDFromEventID(%s) returned 0 NID, was this event stored?", eventID)
|
||||
}
|
||||
|
||||
stateEntries, err := v.LoadStateAtSnapshot(ctx, snapshotNID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return stateEntries, nil
|
||||
}
|
||||
|
||||
// LoadCombinedStateAfterEvents loads a snapshot of the state after each of the events
|
||||
// and combines those snapshots together into a single list. At this point it is
|
||||
// possible to run into duplicate (type, state key) tuples.
|
||||
|
|
|
|||
|
|
@ -166,4 +166,6 @@ type Database interface {
|
|||
GetKnownRooms(ctx context.Context) ([]string, error)
|
||||
// ForgetRoom sets a flag in the membership table, that the user wishes to forget a specific room
|
||||
ForgetRoom(ctx context.Context, userID, roomID string, forget bool) error
|
||||
|
||||
GetHistoryVisibilityState(ctx context.Context, roomInfo *types.RoomInfo, eventID string, domain string) ([]*gomatrixserverlib.Event, error)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -147,14 +147,16 @@ func (s *roomStatements) InsertRoomNID(
|
|||
func (s *roomStatements) SelectRoomInfo(ctx context.Context, txn *sql.Tx, roomID string) (*types.RoomInfo, error) {
|
||||
var info types.RoomInfo
|
||||
var latestNIDs pq.Int64Array
|
||||
var stateSnapshotNID types.StateSnapshotNID
|
||||
stmt := sqlutil.TxStmt(txn, s.selectRoomInfoStmt)
|
||||
err := stmt.QueryRowContext(ctx, roomID).Scan(
|
||||
&info.RoomVersion, &info.RoomNID, &info.StateSnapshotNID, &latestNIDs,
|
||||
&info.RoomVersion, &info.RoomNID, &stateSnapshotNID, &latestNIDs,
|
||||
)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, nil
|
||||
}
|
||||
info.IsStub = len(latestNIDs) == 0
|
||||
info.SetStateSnapshotNID(stateSnapshotNID)
|
||||
info.SetIsStub(len(latestNIDs) == 0)
|
||||
return &info, err
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -72,9 +72,35 @@ const bulkSelectStateBlockNIDsSQL = "" +
|
|||
"SELECT state_snapshot_nid, state_block_nids FROM roomserver_state_snapshots" +
|
||||
" WHERE state_snapshot_nid = ANY($1) ORDER BY state_snapshot_nid ASC"
|
||||
|
||||
// Looks up both the history visibility event and relevant membership events from
|
||||
// a given domain name from a given state snapshot. This is used to optimise the
|
||||
// helpers.CheckServerAllowedToSeeEvent function.
|
||||
// TODO: There's a sequence scan here because of the hash join strategy, which is
|
||||
// probably O(n) on state key entries, so there must be a way to avoid that somehow.
|
||||
// Event type NIDs are:
|
||||
// - 5: m.room.member as per https://github.com/matrix-org/dendrite/blob/c7f7aec4d07d59120d37d5b16a900f6d608a75c4/roomserver/storage/postgres/event_types_table.go#L40
|
||||
// - 7: m.room.history_visibility as per https://github.com/matrix-org/dendrite/blob/c7f7aec4d07d59120d37d5b16a900f6d608a75c4/roomserver/storage/postgres/event_types_table.go#L42
|
||||
const bulkSelectStateForHistoryVisibilitySQL = `
|
||||
SELECT event_nid FROM (
|
||||
SELECT event_nid, event_type_nid, event_state_key_nid FROM roomserver_events
|
||||
WHERE (event_type_nid = 5 OR event_type_nid = 7)
|
||||
AND event_nid = ANY(
|
||||
SELECT UNNEST(event_nids) FROM roomserver_state_block
|
||||
WHERE state_block_nid = ANY(
|
||||
SELECT UNNEST(state_block_nids) FROM roomserver_state_snapshots
|
||||
WHERE state_snapshot_nid = $1
|
||||
)
|
||||
)
|
||||
) AS roomserver_events
|
||||
INNER JOIN roomserver_event_state_keys
|
||||
ON roomserver_events.event_state_key_nid = roomserver_event_state_keys.event_state_key_nid
|
||||
AND (event_type_nid = 7 OR event_state_key LIKE '%:' || $2);
|
||||
`
|
||||
|
||||
type stateSnapshotStatements struct {
|
||||
insertStateStmt *sql.Stmt
|
||||
bulkSelectStateBlockNIDsStmt *sql.Stmt
|
||||
insertStateStmt *sql.Stmt
|
||||
bulkSelectStateBlockNIDsStmt *sql.Stmt
|
||||
bulkSelectStateForHistoryVisibilityStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func CreateStateSnapshotTable(db *sql.DB) error {
|
||||
|
|
@ -88,6 +114,7 @@ func PrepareStateSnapshotTable(db *sql.DB) (tables.StateSnapshot, error) {
|
|||
return s, sqlutil.StatementList{
|
||||
{&s.insertStateStmt, insertStateSQL},
|
||||
{&s.bulkSelectStateBlockNIDsStmt, bulkSelectStateBlockNIDsSQL},
|
||||
{&s.bulkSelectStateForHistoryVisibilityStmt, bulkSelectStateForHistoryVisibilitySQL},
|
||||
}.Prepare(db)
|
||||
}
|
||||
|
||||
|
|
@ -136,3 +163,23 @@ func (s *stateSnapshotStatements) BulkSelectStateBlockNIDs(
|
|||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (s *stateSnapshotStatements) BulkSelectStateForHistoryVisibility(
|
||||
ctx context.Context, txn *sql.Tx, stateSnapshotNID types.StateSnapshotNID, domain string,
|
||||
) ([]types.EventNID, error) {
|
||||
stmt := sqlutil.TxStmt(txn, s.bulkSelectStateForHistoryVisibilityStmt)
|
||||
rows, err := stmt.QueryContext(ctx, stateSnapshotNID, domain)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close() // nolint: errcheck
|
||||
results := make([]types.EventNID, 0, 16)
|
||||
for rows.Next() {
|
||||
var eventNID types.EventNID
|
||||
if err = rows.Scan(&eventNID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, eventNID)
|
||||
}
|
||||
return results, rows.Err()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -229,8 +229,8 @@ func (u *RoomUpdater) SetLatestEvents(
|
|||
// Since it's entirely possible that this types.RoomInfo came from the
|
||||
// cache, we should make sure to update that entry so that the next run
|
||||
// works from live data.
|
||||
u.roomInfo.StateSnapshotNID = currentStateSnapshotNID
|
||||
u.roomInfo.IsStub = false
|
||||
u.roomInfo.SetStateSnapshotNID(currentStateSnapshotNID)
|
||||
u.roomInfo.SetIsStub(false)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -988,6 +988,38 @@ func (d *Database) loadEvent(ctx context.Context, eventID string) *types.Event {
|
|||
return &evs[0]
|
||||
}
|
||||
|
||||
func (d *Database) GetHistoryVisibilityState(ctx context.Context, roomInfo *types.RoomInfo, eventID string, domain string) ([]*gomatrixserverlib.Event, error) {
|
||||
eventStates, err := d.EventsTable.BulkSelectStateAtEventByID(ctx, nil, []string{eventID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stateSnapshotNID := eventStates[0].BeforeStateSnapshotNID
|
||||
if stateSnapshotNID == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
eventNIDs, err := d.StateSnapshotTable.BulkSelectStateForHistoryVisibility(ctx, nil, stateSnapshotNID, domain)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
eventIDs, _ := d.EventsTable.BulkSelectEventID(ctx, nil, eventNIDs)
|
||||
if err != nil {
|
||||
eventIDs = map[types.EventNID]string{}
|
||||
}
|
||||
events := make([]*gomatrixserverlib.Event, 0, len(eventNIDs))
|
||||
for _, eventNID := range eventNIDs {
|
||||
data, err := d.EventJSONTable.BulkSelectEventJSON(ctx, nil, []types.EventNID{eventNID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ev, err := gomatrixserverlib.NewEventFromTrustedJSONWithEventID(eventIDs[eventNID], data[0].EventJSON, false, roomInfo.RoomVersion)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
events = append(events, ev)
|
||||
}
|
||||
return events, nil
|
||||
}
|
||||
|
||||
// GetStateEvent returns the current 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
|
||||
|
|
@ -1000,7 +1032,7 @@ func (d *Database) GetStateEvent(ctx context.Context, roomID, evType, stateKey s
|
|||
return nil, fmt.Errorf("room %s doesn't exist", roomID)
|
||||
}
|
||||
// e.g invited rooms
|
||||
if roomInfo.IsStub {
|
||||
if roomInfo.IsStub() {
|
||||
return nil, nil
|
||||
}
|
||||
eventTypeNID, err := d.EventTypesTable.SelectEventTypeNID(ctx, nil, evType)
|
||||
|
|
@ -1019,7 +1051,7 @@ func (d *Database) GetStateEvent(ctx context.Context, roomID, evType, stateKey s
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entries, err := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID)
|
||||
entries, err := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -1065,7 +1097,7 @@ func (d *Database) GetStateEventsWithEventType(ctx context.Context, roomID, evTy
|
|||
return nil, fmt.Errorf("room %s doesn't exist", roomID)
|
||||
}
|
||||
// e.g invited rooms
|
||||
if roomInfo.IsStub {
|
||||
if roomInfo.IsStub() {
|
||||
return nil, nil
|
||||
}
|
||||
eventTypeNID, err := d.EventTypesTable.SelectEventTypeNID(ctx, nil, evType)
|
||||
|
|
@ -1076,7 +1108,7 @@ func (d *Database) GetStateEventsWithEventType(ctx context.Context, roomID, evTy
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entries, err := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID)
|
||||
entries, err := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -1193,10 +1225,10 @@ func (d *Database) GetBulkStateContent(ctx context.Context, roomIDs []string, tu
|
|||
return nil, fmt.Errorf("GetBulkStateContent: failed to load room info for room %s : %w", roomID, err2)
|
||||
}
|
||||
// for unknown rooms or rooms which we don't have the current state, skip them.
|
||||
if roomInfo == nil || roomInfo.IsStub {
|
||||
if roomInfo == nil || roomInfo.IsStub() {
|
||||
continue
|
||||
}
|
||||
entries, err2 := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID)
|
||||
entries, err2 := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID())
|
||||
if err2 != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to load state for room %s : %w", roomID, err2)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -129,9 +129,10 @@ func (s *roomStatements) SelectRoomIDsWithEvents(ctx context.Context, txn *sql.T
|
|||
func (s *roomStatements) SelectRoomInfo(ctx context.Context, txn *sql.Tx, roomID string) (*types.RoomInfo, error) {
|
||||
var info types.RoomInfo
|
||||
var latestNIDsJSON string
|
||||
var stateSnapshotNID types.StateSnapshotNID
|
||||
stmt := sqlutil.TxStmt(txn, s.selectRoomInfoStmt)
|
||||
err := stmt.QueryRowContext(ctx, roomID).Scan(
|
||||
&info.RoomVersion, &info.RoomNID, &info.StateSnapshotNID, &latestNIDsJSON,
|
||||
&info.RoomVersion, &info.RoomNID, &stateSnapshotNID, &latestNIDsJSON,
|
||||
)
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
|
|
@ -143,7 +144,8 @@ func (s *roomStatements) SelectRoomInfo(ctx context.Context, txn *sql.Tx, roomID
|
|||
if err = json.Unmarshal([]byte(latestNIDsJSON), &latestNIDs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
info.IsStub = len(latestNIDs) == 0
|
||||
info.SetStateSnapshotNID(stateSnapshotNID)
|
||||
info.SetIsStub(len(latestNIDs) == 0)
|
||||
return &info, err
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -140,3 +140,9 @@ func (s *stateSnapshotStatements) BulkSelectStateBlockNIDs(
|
|||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (s *stateSnapshotStatements) BulkSelectStateForHistoryVisibility(
|
||||
ctx context.Context, txn *sql.Tx, stateSnapshotNID types.StateSnapshotNID, domain string,
|
||||
) ([]types.EventNID, error) {
|
||||
return nil, tables.OptimisationNotSupportedError
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,12 +3,15 @@ package tables
|
|||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
|
||||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
|
||||
var OptimisationNotSupportedError = errors.New("optimisation not supported")
|
||||
|
||||
type EventJSONPair struct {
|
||||
EventNID types.EventNID
|
||||
EventJSON []byte
|
||||
|
|
@ -80,6 +83,10 @@ type Rooms interface {
|
|||
type StateSnapshot interface {
|
||||
InsertState(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, stateBlockNIDs types.StateBlockNIDs) (stateNID types.StateSnapshotNID, err error)
|
||||
BulkSelectStateBlockNIDs(ctx context.Context, txn *sql.Tx, stateNIDs []types.StateSnapshotNID) ([]types.StateBlockNIDList, error)
|
||||
// BulkSelectStateForHistoryVisibility is a PostgreSQL-only optimisation for finding
|
||||
// which users are in a room faster than having to load the entire room state. In the
|
||||
// case of SQLite, this will return tables.OptimisationNotSupportedError.
|
||||
BulkSelectStateForHistoryVisibility(ctx context.Context, txn *sql.Tx, stateSnapshotNID types.StateSnapshotNID, domain string) ([]types.EventNID, error)
|
||||
}
|
||||
|
||||
type StateBlock interface {
|
||||
|
|
|
|||
|
|
@ -63,12 +63,12 @@ func TestRoomsTable(t *testing.T) {
|
|||
|
||||
roomInfo, err := tab.SelectRoomInfo(ctx, nil, room.ID)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, &types.RoomInfo{
|
||||
RoomNID: wantRoomNID,
|
||||
RoomVersion: room.Version,
|
||||
StateSnapshotNID: 0,
|
||||
IsStub: true, // there are no latestEventNIDs
|
||||
}, roomInfo)
|
||||
expected := &types.RoomInfo{
|
||||
RoomNID: wantRoomNID,
|
||||
RoomVersion: room.Version,
|
||||
}
|
||||
expected.SetIsStub(true) // there are no latestEventNIDs
|
||||
assert.Equal(t, expected, roomInfo)
|
||||
|
||||
roomInfo, err = tab.SelectRoomInfo(ctx, nil, "!doesnotexist:localhost")
|
||||
assert.NoError(t, err)
|
||||
|
|
@ -103,12 +103,12 @@ func TestRoomsTable(t *testing.T) {
|
|||
|
||||
roomInfo, err = tab.SelectRoomInfo(ctx, nil, room.ID)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, &types.RoomInfo{
|
||||
RoomNID: wantRoomNID,
|
||||
RoomVersion: room.Version,
|
||||
StateSnapshotNID: 1,
|
||||
IsStub: false,
|
||||
}, roomInfo)
|
||||
expected = &types.RoomInfo{
|
||||
RoomNID: wantRoomNID,
|
||||
RoomVersion: room.Version,
|
||||
}
|
||||
expected.SetStateSnapshotNID(1)
|
||||
assert.Equal(t, expected, roomInfo)
|
||||
|
||||
eventNIDs, snapshotNID, err := tab.SelectLatestEventNIDs(ctx, nil, wantRoomNID)
|
||||
assert.NoError(t, err)
|
||||
|
|
|
|||
|
|
@ -23,6 +23,15 @@ func mustCreateStateSnapshotTable(t *testing.T, dbType test.DBType) (tab tables.
|
|||
assert.NoError(t, err)
|
||||
switch dbType {
|
||||
case test.DBTypePostgres:
|
||||
// for the PostgreSQL history visibility optimisation to work,
|
||||
// we also need some other tables to exist
|
||||
err = postgres.CreateEventStateKeysTable(db)
|
||||
assert.NoError(t, err)
|
||||
err = postgres.CreateEventsTable(db)
|
||||
assert.NoError(t, err)
|
||||
err = postgres.CreateStateBlockTable(db)
|
||||
assert.NoError(t, err)
|
||||
// ... and then the snapshot table itself
|
||||
err = postgres.CreateStateSnapshotTable(db)
|
||||
assert.NoError(t, err)
|
||||
tab, err = postgres.PrepareStateSnapshotTable(db)
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"encoding/json"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
|
|
@ -279,8 +280,33 @@ func (e RejectedError) Error() string { return string(e) }
|
|||
|
||||
// RoomInfo contains metadata about a room
|
||||
type RoomInfo struct {
|
||||
mu sync.RWMutex
|
||||
RoomNID RoomNID
|
||||
RoomVersion gomatrixserverlib.RoomVersion
|
||||
StateSnapshotNID StateSnapshotNID
|
||||
IsStub bool
|
||||
stateSnapshotNID StateSnapshotNID
|
||||
isStub bool
|
||||
}
|
||||
|
||||
func (r *RoomInfo) StateSnapshotNID() StateSnapshotNID {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
return r.stateSnapshotNID
|
||||
}
|
||||
|
||||
func (r *RoomInfo) IsStub() bool {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
return r.isStub
|
||||
}
|
||||
|
||||
func (r *RoomInfo) SetStateSnapshotNID(nid StateSnapshotNID) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.stateSnapshotNID = nid
|
||||
}
|
||||
|
||||
func (r *RoomInfo) SetIsStub(isStub bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.isStub = isStub
|
||||
}
|
||||
|
|
|
|||
|
|
@ -369,6 +369,25 @@ func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationCli
|
|||
return client
|
||||
}
|
||||
|
||||
func (b *BaseDendrite) configureHTTPErrors() {
|
||||
notAllowedHandler := func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
_, _ = w.Write([]byte(fmt.Sprintf("405 %s not allowed on this endpoint", r.Method)))
|
||||
}
|
||||
|
||||
notFoundCORSHandler := httputil.WrapHandlerInCORS(http.NotFoundHandler())
|
||||
notAllowedCORSHandler := httputil.WrapHandlerInCORS(http.HandlerFunc(notAllowedHandler))
|
||||
|
||||
for _, router := range []*mux.Router{
|
||||
b.PublicClientAPIMux, b.PublicMediaAPIMux,
|
||||
b.DendriteAdminMux, b.SynapseAdminMux,
|
||||
b.PublicWellKnownAPIMux,
|
||||
} {
|
||||
router.NotFoundHandler = notFoundCORSHandler
|
||||
router.MethodNotAllowedHandler = notAllowedCORSHandler
|
||||
}
|
||||
}
|
||||
|
||||
// SetupAndServeHTTP sets up the HTTP server to serve endpoints registered on
|
||||
// ApiMux under /api/ and adds a prometheus handler under /metrics.
|
||||
func (b *BaseDendrite) SetupAndServeHTTP(
|
||||
|
|
@ -409,6 +428,8 @@ func (b *BaseDendrite) SetupAndServeHTTP(
|
|||
}
|
||||
}
|
||||
|
||||
b.configureHTTPErrors()
|
||||
|
||||
internalRouter.PathPrefix(httputil.InternalPathPrefix).Handler(b.InternalAPIMux)
|
||||
if b.Cfg.Global.Metrics.Enabled {
|
||||
internalRouter.Handle("/metrics", httputil.WrapHandlerInBasicAuth(promhttp.Handler(), b.Cfg.Global.Metrics.BasicAuth))
|
||||
|
|
|
|||
|
|
@ -109,12 +109,11 @@ func (p *PDUStreamProvider) CompleteSync(
|
|||
p.queue(func() {
|
||||
defer reqWaitGroup.Done()
|
||||
|
||||
var jr *types.JoinResponse
|
||||
jr, err = p.getJoinResponseForCompleteSync(
|
||||
jr, jerr := p.getJoinResponseForCompleteSync(
|
||||
ctx, roomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
|
||||
)
|
||||
if err != nil {
|
||||
req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
|
||||
if jerr != nil {
|
||||
req.Log.WithError(jerr).Error("p.getJoinResponseForCompleteSync failed")
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -262,9 +261,9 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
|
|||
var pos types.StreamPosition
|
||||
if _, pos, err = p.DB.PositionInTopology(ctx, mostRecentEventID); err == nil {
|
||||
switch {
|
||||
case r.Backwards && pos > latestPosition:
|
||||
case r.Backwards && pos < latestPosition:
|
||||
fallthrough
|
||||
case !r.Backwards && pos < latestPosition:
|
||||
case !r.Backwards && pos > latestPosition:
|
||||
latestPosition = pos
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,11 +30,13 @@ import (
|
|||
|
||||
"github.com/matrix-org/dendrite/appservice/types"
|
||||
"github.com/matrix-org/dendrite/clientapi/userutil"
|
||||
"github.com/matrix-org/dendrite/internal/eventutil"
|
||||
"github.com/matrix-org/dendrite/internal/pushrules"
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
keyapi "github.com/matrix-org/dendrite/keyserver/api"
|
||||
rsapi "github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/setup/config"
|
||||
synctypes "github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/dendrite/userapi/api"
|
||||
"github.com/matrix-org/dendrite/userapi/producers"
|
||||
"github.com/matrix-org/dendrite/userapi/storage"
|
||||
|
|
@ -64,7 +66,24 @@ func (a *UserInternalAPI) InputAccountData(ctx context.Context, req *api.InputAc
|
|||
if req.DataType == "" {
|
||||
return fmt.Errorf("data type must not be empty")
|
||||
}
|
||||
return a.DB.SaveAccountData(ctx, local, req.RoomID, req.DataType, req.AccountData)
|
||||
if err := a.DB.SaveAccountData(ctx, local, req.RoomID, req.DataType, req.AccountData); err != nil {
|
||||
util.GetLogger(ctx).WithError(err).Error("a.DB.SaveAccountData failed")
|
||||
return fmt.Errorf("failed to save account data: %w", err)
|
||||
}
|
||||
var ignoredUsers *synctypes.IgnoredUsers
|
||||
if req.DataType == "m.ignored_user_list" {
|
||||
ignoredUsers = &synctypes.IgnoredUsers{}
|
||||
_ = json.Unmarshal(req.AccountData, ignoredUsers)
|
||||
}
|
||||
if err := a.SyncProducer.SendAccountData(req.UserID, eventutil.AccountData{
|
||||
RoomID: req.RoomID,
|
||||
Type: req.DataType,
|
||||
IgnoredUsers: ignoredUsers,
|
||||
}); err != nil {
|
||||
util.GetLogger(ctx).WithError(err).Error("a.SyncProducer.SendAccountData failed")
|
||||
return fmt.Errorf("failed to send account data to output: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *UserInternalAPI) PerformAccountCreation(ctx context.Context, req *api.PerformAccountCreationRequest, res *api.PerformAccountCreationResponse) error {
|
||||
|
|
@ -93,7 +112,9 @@ func (a *UserInternalAPI) PerformAccountCreation(ctx context.Context, req *api.P
|
|||
}
|
||||
|
||||
// Inform the SyncAPI about the newly created push_rules
|
||||
if err = a.SyncProducer.SendAccountData(acc.UserID, "", "m.push_rules"); err != nil {
|
||||
if err = a.SyncProducer.SendAccountData(acc.UserID, eventutil.AccountData{
|
||||
Type: "m.push_rules",
|
||||
}); err != nil {
|
||||
util.GetLogger(ctx).WithFields(logrus.Fields{
|
||||
"user_id": acc.UserID,
|
||||
}).WithError(err).Warn("failed to send account data to the SyncAPI")
|
||||
|
|
@ -732,11 +753,11 @@ func (a *UserInternalAPI) PerformPushRulesPut(
|
|||
if err := a.InputAccountData(ctx, &userReq, &userRes); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := a.SyncProducer.SendAccountData(req.UserID, "" /* roomID */, pushRulesAccountDataType); err != nil {
|
||||
if err := a.SyncProducer.SendAccountData(req.UserID, eventutil.AccountData{
|
||||
Type: pushRulesAccountDataType,
|
||||
}); err != nil {
|
||||
util.GetLogger(ctx).WithError(err).Errorf("syncProducer.SendData failed")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ func NewSyncAPI(db storage.Database, js JetStreamPublisher, clientDataTopic stri
|
|||
}
|
||||
|
||||
// SendAccountData sends account data to the Sync API server.
|
||||
func (p *SyncAPI) SendAccountData(userID string, roomID string, dataType string) error {
|
||||
func (p *SyncAPI) SendAccountData(userID string, data eventutil.AccountData) error {
|
||||
m := &nats.Msg{
|
||||
Subject: p.clientDataTopic,
|
||||
Header: nats.Header{},
|
||||
|
|
@ -42,18 +42,15 @@ func (p *SyncAPI) SendAccountData(userID string, roomID string, dataType string)
|
|||
m.Header.Set(jetstream.UserID, userID)
|
||||
|
||||
var err error
|
||||
m.Data, err = json.Marshal(eventutil.AccountData{
|
||||
RoomID: roomID,
|
||||
Type: dataType,
|
||||
})
|
||||
m.Data, err = json.Marshal(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"user_id": userID,
|
||||
"room_id": roomID,
|
||||
"data_type": dataType,
|
||||
"room_id": data.RoomID,
|
||||
"data_type": data.Type,
|
||||
}).Tracef("Producing to topic '%s'", p.clientDataTopic)
|
||||
|
||||
_, err = p.producer.PublishMsg(m)
|
||||
|
|
|
|||
Loading…
Reference in a new issue