Merge branch 'develop' into matrix-org-hotfixes
This commit is contained in:
commit
67acd1aa1b
|
@ -1 +1 @@
|
|||
Better checks on newsfragments
|
||||
Better checks on newsfragments.
|
||||
|
|
1
changelog.d/4746.feature
Normal file
1
changelog.d/4746.feature
Normal file
|
@ -0,0 +1 @@
|
|||
Prevent showing rooms to other servers that were set to not federate.
|
1
changelog.d/4748.misc
Normal file
1
changelog.d/4748.misc
Normal file
|
@ -0,0 +1 @@
|
|||
Improve replication performance by reducing cache invalidation traffic.
|
1
changelog.d/4750.misc
Normal file
1
changelog.d/4750.misc
Normal file
|
@ -0,0 +1 @@
|
|||
Better checks on newsfragments.
|
1
changelog.d/4753.misc
Normal file
1
changelog.d/4753.misc
Normal file
|
@ -0,0 +1 @@
|
|||
Add database version to phonehome stats.
|
|
@ -79,7 +79,7 @@ Let's assume that we expect clients to connect to our server at
|
|||
SSLEngine on
|
||||
ServerName example.com;
|
||||
|
||||
<Location />
|
||||
<Location /_matrix>
|
||||
ProxyPass http://127.0.0.1:8008/_matrix nocanon
|
||||
ProxyPassReverse http://127.0.0.1:8008/_matrix
|
||||
</Location>
|
||||
|
|
|
@ -6,7 +6,8 @@
|
|||
set -e
|
||||
|
||||
# make sure that origin/develop is up to date
|
||||
git fetch origin develop
|
||||
git remote set-branches --add origin develop
|
||||
git fetch --depth=1 origin develop
|
||||
|
||||
UPSTREAM=origin/develop
|
||||
|
||||
|
@ -25,11 +26,15 @@ if git diff --name-only $UPSTREAM... | grep -qv '^develop/'; then
|
|||
tox -e check-newsfragment
|
||||
fi
|
||||
|
||||
echo
|
||||
echo "--------------------------"
|
||||
echo
|
||||
|
||||
# check that any new newsfiles on this branch end with a full stop.
|
||||
for f in git diff --name-only $UPSTREAM... -- changelog.d; do
|
||||
for f in `git diff --name-only $UPSTREAM... -- changelog.d`; do
|
||||
lastchar=`tr -d '\n' < $f | tail -c 1`
|
||||
if [ $lastchar != '.' ]; then
|
||||
echo "Newsfragment $f does not end with a '.'" >&2
|
||||
echo -e "\e[31mERROR: newsfragment $f does not end with a '.'\e[39m" >&2
|
||||
exit 1
|
||||
fi
|
||||
done
|
||||
|
|
|
@ -555,6 +555,9 @@ def run(hs):
|
|||
stats["memory_rss"] += process.memory_info().rss
|
||||
stats["cpu_average"] += int(process.cpu_percent(interval=None))
|
||||
|
||||
stats["database_engine"] = hs.get_datastore().database_engine_name
|
||||
stats["database_server_version"] = hs.get_datastore().get_server_version()
|
||||
|
||||
logger.info("Reporting stats to matrix.org: %s" % (stats,))
|
||||
try:
|
||||
yield hs.get_simple_http_client().put_json(
|
||||
|
|
|
@ -736,7 +736,8 @@ class PublicRoomList(BaseFederationServlet):
|
|||
|
||||
data = yield self.handler.get_local_public_room_list(
|
||||
limit, since_token,
|
||||
network_tuple=network_tuple
|
||||
network_tuple=network_tuple,
|
||||
from_federation=True,
|
||||
)
|
||||
defer.returnValue((200, data))
|
||||
|
||||
|
|
|
@ -113,8 +113,7 @@ class GroupsServerHandler(object):
|
|||
room_id = room_entry["room_id"]
|
||||
joined_users = yield self.store.get_users_in_room(room_id)
|
||||
entry = yield self.room_list_handler.generate_room_entry(
|
||||
room_id, len(joined_users),
|
||||
with_alias=False, allow_private=True,
|
||||
room_id, len(joined_users), with_alias=False, allow_private=True,
|
||||
)
|
||||
entry = dict(entry) # so we don't change whats cached
|
||||
entry.pop("room_id", None)
|
||||
|
@ -544,8 +543,7 @@ class GroupsServerHandler(object):
|
|||
|
||||
joined_users = yield self.store.get_users_in_room(room_id)
|
||||
entry = yield self.room_list_handler.generate_room_entry(
|
||||
room_id, len(joined_users),
|
||||
with_alias=False, allow_private=True,
|
||||
room_id, len(joined_users), with_alias=False, allow_private=True,
|
||||
)
|
||||
|
||||
if not entry:
|
||||
|
|
|
@ -53,16 +53,17 @@ class RoomListHandler(BaseHandler):
|
|||
|
||||
def get_local_public_room_list(self, limit=None, since_token=None,
|
||||
search_filter=None,
|
||||
network_tuple=EMPTY_THIRD_PARTY_ID,):
|
||||
network_tuple=EMPTY_THIRD_PARTY_ID,
|
||||
from_federation=False):
|
||||
"""Generate a local public room list.
|
||||
|
||||
There are multiple different lists: the main one plus one per third
|
||||
party network. A client can ask for a specific list or to return all.
|
||||
|
||||
Args:
|
||||
limit (int)
|
||||
since_token (str)
|
||||
search_filter (dict)
|
||||
limit (int|None)
|
||||
since_token (str|None)
|
||||
search_filter (dict|None)
|
||||
network_tuple (ThirdPartyInstanceID): Which public list to use.
|
||||
This can be (None, None) to indicate the main list, or a particular
|
||||
appservice and network id to use an appservice specific one.
|
||||
|
@ -90,14 +91,30 @@ class RoomListHandler(BaseHandler):
|
|||
return self.response_cache.wrap(
|
||||
key,
|
||||
self._get_public_room_list,
|
||||
limit, since_token, network_tuple=network_tuple,
|
||||
limit, since_token,
|
||||
network_tuple=network_tuple, from_federation=from_federation,
|
||||
)
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def _get_public_room_list(self, limit=None, since_token=None,
|
||||
search_filter=None,
|
||||
network_tuple=EMPTY_THIRD_PARTY_ID,
|
||||
from_federation=False,
|
||||
timeout=None,):
|
||||
"""Generate a public room list.
|
||||
Args:
|
||||
limit (int|None): Maximum amount of rooms to return.
|
||||
since_token (str|None)
|
||||
search_filter (dict|None): Dictionary to filter rooms by.
|
||||
network_tuple (ThirdPartyInstanceID): Which public list to use.
|
||||
This can be (None, None) to indicate the main list, or a particular
|
||||
appservice and network id to use an appservice specific one.
|
||||
Setting to None returns all public rooms across all lists.
|
||||
from_federation (bool): Whether this request originated from a
|
||||
federating server or a client. Used for room filtering.
|
||||
timeout (int|None): Amount of seconds to wait for a response before
|
||||
timing out.
|
||||
"""
|
||||
if since_token and since_token != "END":
|
||||
since_token = RoomListNextBatch.from_token(since_token)
|
||||
else:
|
||||
|
@ -220,7 +237,8 @@ class RoomListHandler(BaseHandler):
|
|||
yield concurrently_execute(
|
||||
lambda r: self._append_room_entry_to_chunk(
|
||||
r, rooms_to_num_joined[r],
|
||||
chunk, limit, search_filter
|
||||
chunk, limit, search_filter,
|
||||
from_federation=from_federation,
|
||||
),
|
||||
batch, 5,
|
||||
)
|
||||
|
@ -291,23 +309,51 @@ class RoomListHandler(BaseHandler):
|
|||
|
||||
@defer.inlineCallbacks
|
||||
def _append_room_entry_to_chunk(self, room_id, num_joined_users, chunk, limit,
|
||||
search_filter):
|
||||
search_filter, from_federation=False):
|
||||
"""Generate the entry for a room in the public room list and append it
|
||||
to the `chunk` if it matches the search filter
|
||||
|
||||
Args:
|
||||
room_id (str): The ID of the room.
|
||||
num_joined_users (int): The number of joined users in the room.
|
||||
chunk (list)
|
||||
limit (int|None): Maximum amount of rooms to display. Function will
|
||||
return if length of chunk is greater than limit + 1.
|
||||
search_filter (dict|None)
|
||||
from_federation (bool): Whether this request originated from a
|
||||
federating server or a client. Used for room filtering.
|
||||
"""
|
||||
if limit and len(chunk) > limit + 1:
|
||||
# We've already got enough, so lets just drop it.
|
||||
return
|
||||
|
||||
result = yield self.generate_room_entry(room_id, num_joined_users)
|
||||
if not result:
|
||||
return
|
||||
|
||||
if result and _matches_room_entry(result, search_filter):
|
||||
if from_federation and not result.get("m.federate", True):
|
||||
# This is a room that other servers cannot join. Do not show them
|
||||
# this room.
|
||||
return
|
||||
|
||||
if _matches_room_entry(result, search_filter):
|
||||
chunk.append(result)
|
||||
|
||||
@cachedInlineCallbacks(num_args=1, cache_context=True)
|
||||
def generate_room_entry(self, room_id, num_joined_users, cache_context,
|
||||
with_alias=True, allow_private=False):
|
||||
"""Returns the entry for a room
|
||||
|
||||
Args:
|
||||
room_id (str): The room's ID.
|
||||
num_joined_users (int): Number of users in the room.
|
||||
cache_context: Information for cached responses.
|
||||
with_alias (bool): Whether to return the room's aliases in the result.
|
||||
allow_private (bool): Whether invite-only rooms should be shown.
|
||||
|
||||
Returns:
|
||||
Deferred[dict|None]: Returns a room entry as a dictionary, or None if this
|
||||
room was determined not to be shown publicly.
|
||||
"""
|
||||
result = {
|
||||
"room_id": room_id,
|
||||
|
@ -321,6 +367,7 @@ class RoomListHandler(BaseHandler):
|
|||
event_map = yield self.store.get_events([
|
||||
event_id for key, event_id in iteritems(current_state_ids)
|
||||
if key[0] in (
|
||||
EventTypes.Create,
|
||||
EventTypes.JoinRules,
|
||||
EventTypes.Name,
|
||||
EventTypes.Topic,
|
||||
|
@ -337,12 +384,17 @@ class RoomListHandler(BaseHandler):
|
|||
}
|
||||
|
||||
# Double check that this is actually a public room.
|
||||
|
||||
join_rules_event = current_state.get((EventTypes.JoinRules, ""))
|
||||
if join_rules_event:
|
||||
join_rule = join_rules_event.content.get("join_rule", None)
|
||||
if not allow_private and join_rule and join_rule != JoinRules.PUBLIC:
|
||||
defer.returnValue(None)
|
||||
|
||||
# Return whether this room is open to federation users or not
|
||||
create_event = current_state.get((EventTypes.Create, ""))
|
||||
result["m.federate"] = create_event.content.get("m.federate", True)
|
||||
|
||||
if with_alias:
|
||||
aliases = yield self.store.get_aliases_for_room(
|
||||
room_id, on_invalidate=cache_context.invalidate
|
||||
|
|
|
@ -270,7 +270,17 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
|
|||
if "\n" in string:
|
||||
raise Exception("Unexpected newline in command: %r", string)
|
||||
|
||||
self.sendLine(string.encode("utf-8"))
|
||||
encoded_string = string.encode("utf-8")
|
||||
|
||||
if len(encoded_string) > self.MAX_LENGTH:
|
||||
raise Exception(
|
||||
"Failed to send command %s as too long (%d > %d)" % (
|
||||
cmd.NAME,
|
||||
len(encoded_string), self.MAX_LENGTH,
|
||||
)
|
||||
)
|
||||
|
||||
self.sendLine(encoded_string)
|
||||
|
||||
self.last_sent_command = self.clock.time_msec()
|
||||
|
||||
|
@ -365,6 +375,11 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
|
|||
def id(self):
|
||||
return "%s-%s" % (self.name, self.conn_id)
|
||||
|
||||
def lineLengthExceeded(self, line):
|
||||
"""Called when we receive a line that is above the maximum line length
|
||||
"""
|
||||
self.send_error("Line length exceeded")
|
||||
|
||||
|
||||
class ServerReplicationStreamProtocol(BaseReplicationStreamProtocol):
|
||||
VALID_INBOUND_COMMANDS = VALID_CLIENT_COMMANDS
|
||||
|
|
|
@ -30,6 +30,7 @@ from synapse.api.errors import StoreError
|
|||
from synapse.metrics.background_process_metrics import run_as_background_process
|
||||
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
|
||||
from synapse.types import get_domain_from_id
|
||||
from synapse.util import batch_iter
|
||||
from synapse.util.caches.descriptors import Cache
|
||||
from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
|
||||
from synapse.util.stringutils import exception_to_unicode
|
||||
|
@ -1327,10 +1328,16 @@ class SQLBaseStore(object):
|
|||
"""
|
||||
txn.call_after(self._invalidate_state_caches, room_id, members_changed)
|
||||
|
||||
keys = itertools.chain([room_id], members_changed)
|
||||
self._send_invalidation_to_replication(
|
||||
txn, _CURRENT_STATE_CACHE_NAME, keys,
|
||||
)
|
||||
# We need to be careful that the size of the `members_changed` list
|
||||
# isn't so large that it causes problems sending over replication, so we
|
||||
# send them in chunks.
|
||||
# Max line length is 16K, and max user ID length is 255, so 50 should
|
||||
# be safe.
|
||||
for chunk in batch_iter(members_changed, 50):
|
||||
keys = itertools.chain([room_id], chunk)
|
||||
self._send_invalidation_to_replication(
|
||||
txn, _CURRENT_STATE_CACHE_NAME, keys,
|
||||
)
|
||||
|
||||
def _invalidate_state_caches(self, room_id, members_changed):
|
||||
"""Invalidates caches that are based on the current state, but does
|
||||
|
@ -1596,6 +1603,14 @@ class SQLBaseStore(object):
|
|||
|
||||
return cls.cursor_to_dict(txn)
|
||||
|
||||
@property
|
||||
def database_engine_name(self):
|
||||
return self.database_engine.module.__name__
|
||||
|
||||
def get_server_version(self):
|
||||
"""Returns a string describing the server version number"""
|
||||
return self.database_engine.server_version
|
||||
|
||||
|
||||
class _RollbackButIsFineException(Exception):
|
||||
""" This exception is used to rollback a transaction without implying
|
||||
|
|
|
@ -23,6 +23,7 @@ class PostgresEngine(object):
|
|||
self.module = database_module
|
||||
self.module.extensions.register_type(self.module.extensions.UNICODE)
|
||||
self.synchronous_commit = database_config.get("synchronous_commit", True)
|
||||
self._version = None # unknown as yet
|
||||
|
||||
def check_database(self, txn):
|
||||
txn.execute("SHOW SERVER_ENCODING")
|
||||
|
@ -87,3 +88,27 @@ class PostgresEngine(object):
|
|||
"""
|
||||
txn.execute("SELECT nextval('state_group_id_seq')")
|
||||
return txn.fetchone()[0]
|
||||
|
||||
@property
|
||||
def server_version(self):
|
||||
"""Returns a string giving the server version. For example: '8.1.5'
|
||||
|
||||
Returns:
|
||||
string
|
||||
"""
|
||||
# note that this is a bit of a hack because it relies on on_new_connection
|
||||
# having been called at least once. Still, that should be a safe bet here.
|
||||
numver = self._version
|
||||
assert numver is not None
|
||||
|
||||
# https://www.postgresql.org/docs/current/libpq-status.html#LIBPQ-PQSERVERVERSION
|
||||
if numver >= 100000:
|
||||
return "%i.%i" % (
|
||||
numver / 10000, numver % 10000,
|
||||
)
|
||||
else:
|
||||
return "%i.%i.%i" % (
|
||||
numver / 10000,
|
||||
(numver % 10000) / 100,
|
||||
numver % 100,
|
||||
)
|
||||
|
|
|
@ -70,6 +70,15 @@ class Sqlite3Engine(object):
|
|||
self._current_state_group_id += 1
|
||||
return self._current_state_group_id
|
||||
|
||||
@property
|
||||
def server_version(self):
|
||||
"""Gets a string giving the server version. For example: '3.22.0'
|
||||
|
||||
Returns:
|
||||
string
|
||||
"""
|
||||
return "%i.%i.%i" % self.module.sqlite_version_info
|
||||
|
||||
|
||||
# Following functions taken from: https://github.com/coleifer/peewee
|
||||
|
||||
|
|
Loading…
Reference in a new issue