This repository has been archived by the owner on Apr 26, 2024. It is now read-only.
-
-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Move catchup of replication streams to worker. #7024
Merged
Merged
Changes from 7 commits
Commits
Show all changes
17 commits
Select commit
Hold shift + click to select a range
7233d38
Move stream fetch DB queries to worker stores.
erikjohnston 811d2ec
Don't panic if streams get behind.
erikjohnston ba90596
Add ability to catchup on stream by talking to master.
erikjohnston 1f83255
Move stream catchup to workers.
erikjohnston 8734b75
Remove unused token param from REPLICATE cmd
erikjohnston 32c6568
Always subscribe to all streams.
erikjohnston 259cdff
Newsfile
erikjohnston 4f2a803
Merge branch 'develop' of github.com:matrix-org/synapse into erikj/ca…
erikjohnston a2070a2
Remove unused 'stream' param of REPLICATE and update docs
erikjohnston ba1a8be
Review comments
erikjohnston 3204b0e
Handle connection closing under us
erikjohnston 2380e40
Remove import loop
erikjohnston e4c5b1d
Review comments
erikjohnston 309aee4
Move calling http replication out of base stream
erikjohnston bd64b8f
Fixup push rules stream
erikjohnston f8038f4
Fix HTTP update_function
erikjohnston 309c7eb
Add some type aliases
erikjohnston File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1 @@ | ||
Move catchup of replication streams logic to worker. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,65 @@ | ||
# -*- coding: utf-8 -*- | ||
# Copyright 2020 The Matrix.org Foundation C.I.C. | ||
# | ||
# Licensed under the Apache License, Version 2.0 (the "License"); | ||
# you may not use this file except in compliance with the License. | ||
# You may obtain a copy of the License at | ||
# | ||
# http://www.apache.org/licenses/LICENSE-2.0 | ||
# | ||
# Unless required by applicable law or agreed to in writing, software | ||
# distributed under the License is distributed on an "AS IS" BASIS, | ||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
# See the License for the specific language governing permissions and | ||
# limitations under the License. | ||
|
||
import logging | ||
|
||
from synapse.api.errors import SynapseError | ||
from synapse.http.servlet import parse_integer | ||
from synapse.replication.http._base import ReplicationEndpoint | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
class ReplicationGetStreamUpdates(ReplicationEndpoint): | ||
"""Fetches stream updates from a server. Used for streams not persisted to | ||
the database, e.g. typing notifications. | ||
""" | ||
erikjohnston marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
NAME = "get_repl_stream_updates" | ||
PATH_ARGS = ("stream_name",) | ||
METHOD = "GET" | ||
|
||
def __init__(self, hs): | ||
super(ReplicationGetStreamUpdates, self).__init__(hs) | ||
erikjohnston marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
from synapse.replication.tcp.streams import STREAMS_MAP | ||
richvdh marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
self.streams = {stream.NAME: stream(hs) for stream in STREAMS_MAP.values()} | ||
|
||
@staticmethod | ||
def _serialize_payload(stream_name, from_token, upto_token, limit): | ||
return {"from_token": from_token, "upto_token": upto_token, "limit": limit} | ||
|
||
async def _handle_request(self, request, stream_name): | ||
stream = self.streams.get(stream_name) | ||
if stream is None: | ||
raise SynapseError(400, "Unknown stream") | ||
|
||
from_token = parse_integer(request, "from_token", required=True) | ||
upto_token = parse_integer(request, "upto_token", required=True) | ||
limit = parse_integer(request, "limit", required=True) | ||
|
||
updates, upto_token, limited = await stream.get_updates_since( | ||
from_token, upto_token, limit | ||
) | ||
|
||
return ( | ||
200, | ||
{"updates": updates, "upto_token": upto_token, "limited": limited}, | ||
) | ||
|
||
|
||
def register_servlets(hs, http_server): | ||
ReplicationGetStreamUpdates(hs).register(http_server) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
where is this used?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This gets used by the
TypingStream.current_token
, which now gets called on workers (this is the equivalent of calling a slaved ID gencurrent_token
function).