mirror of
https://git.anonymousland.org/anonymousland/synapse-product.git
synced 2024-09-08 07:12:04 +00:00
Merge pull request #676 from matrix-org/markjh/replicate_stateIII
Add replication streams for ex outliers and current state resets
This commit is contained in:
commit
62e395f0e3
@ -204,17 +204,28 @@ class ReplicationResource(Resource):
|
|||||||
request_events = current_token.events
|
request_events = current_token.events
|
||||||
if request_backfill is None:
|
if request_backfill is None:
|
||||||
request_backfill = current_token.backfill
|
request_backfill = current_token.backfill
|
||||||
events_rows, backfill_rows = yield self.store.get_all_new_events(
|
res = yield self.store.get_all_new_events(
|
||||||
request_backfill, request_events,
|
request_backfill, request_events,
|
||||||
current_token.backfill, current_token.events,
|
current_token.backfill, current_token.events,
|
||||||
limit
|
limit
|
||||||
)
|
)
|
||||||
writer.write_header_and_rows("events", events_rows, (
|
writer.write_header_and_rows("events", res.new_forward_events, (
|
||||||
"position", "internal", "json", "state_group"
|
"position", "internal", "json", "state_group"
|
||||||
))
|
))
|
||||||
writer.write_header_and_rows("backfill", backfill_rows, (
|
writer.write_header_and_rows("backfill", res.new_backfill_events, (
|
||||||
"position", "internal", "json", "state_group"
|
"position", "internal", "json", "state_group"
|
||||||
))
|
))
|
||||||
|
writer.write_header_and_rows(
|
||||||
|
"forward_ex_outliers", res.forward_ex_outliers,
|
||||||
|
("position", "event_id", "state_group")
|
||||||
|
)
|
||||||
|
writer.write_header_and_rows(
|
||||||
|
"backward_ex_outliers", res.backward_ex_outliers,
|
||||||
|
("position", "event_id", "state_group")
|
||||||
|
)
|
||||||
|
writer.write_header_and_rows(
|
||||||
|
"state_resets", res.state_resets, ("position",)
|
||||||
|
)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def presence(self, writer, current_token):
|
def presence(self, writer, current_token):
|
||||||
|
@ -25,7 +25,7 @@ from synapse.api.constants import EventTypes
|
|||||||
|
|
||||||
from canonicaljson import encode_canonical_json
|
from canonicaljson import encode_canonical_json
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
|
from collections import namedtuple
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import math
|
import math
|
||||||
@ -205,6 +205,15 @@ class EventsStore(SQLBaseStore):
|
|||||||
txn.call_after(self.get_joined_hosts_for_room.invalidate, (event.room_id,))
|
txn.call_after(self.get_joined_hosts_for_room.invalidate, (event.room_id,))
|
||||||
txn.call_after(self.get_room_name_and_aliases, event.room_id)
|
txn.call_after(self.get_room_name_and_aliases, event.room_id)
|
||||||
|
|
||||||
|
# Add an entry to the current_state_resets table to record the point
|
||||||
|
# where we clobbered the current state
|
||||||
|
stream_order = event.internal_metadata.stream_ordering
|
||||||
|
self._simple_insert_txn(
|
||||||
|
txn,
|
||||||
|
table="current_state_resets",
|
||||||
|
values={"event_stream_ordering": stream_order}
|
||||||
|
)
|
||||||
|
|
||||||
self._simple_delete_txn(
|
self._simple_delete_txn(
|
||||||
txn,
|
txn,
|
||||||
table="current_state_events",
|
table="current_state_events",
|
||||||
@ -314,6 +323,18 @@ class EventsStore(SQLBaseStore):
|
|||||||
(metadata_json, event.event_id,)
|
(metadata_json, event.event_id,)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
stream_order = event.internal_metadata.stream_ordering
|
||||||
|
state_group_id = context.state_group or context.new_state_group_id
|
||||||
|
self._simple_insert_txn(
|
||||||
|
txn,
|
||||||
|
table="ex_outlier_stream",
|
||||||
|
values={
|
||||||
|
"event_stream_ordering": stream_order,
|
||||||
|
"event_id": event.event_id,
|
||||||
|
"state_group": state_group_id,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
sql = (
|
sql = (
|
||||||
"UPDATE events SET outlier = ?"
|
"UPDATE events SET outlier = ?"
|
||||||
" WHERE event_id = ?"
|
" WHERE event_id = ?"
|
||||||
@ -1110,8 +1131,34 @@ class EventsStore(SQLBaseStore):
|
|||||||
if last_forward_id != current_forward_id:
|
if last_forward_id != current_forward_id:
|
||||||
txn.execute(sql, (last_forward_id, current_forward_id, limit))
|
txn.execute(sql, (last_forward_id, current_forward_id, limit))
|
||||||
new_forward_events = txn.fetchall()
|
new_forward_events = txn.fetchall()
|
||||||
|
|
||||||
|
if len(new_forward_events) == limit:
|
||||||
|
upper_bound = new_forward_events[-1][0]
|
||||||
|
else:
|
||||||
|
upper_bound = current_forward_id
|
||||||
|
|
||||||
|
sql = (
|
||||||
|
"SELECT -event_stream_ordering FROM current_state_resets"
|
||||||
|
" WHERE ? < event_stream_ordering"
|
||||||
|
" AND event_stream_ordering <= ?"
|
||||||
|
" ORDER BY event_stream_ordering ASC"
|
||||||
|
)
|
||||||
|
txn.execute(sql, (last_forward_id, upper_bound))
|
||||||
|
state_resets = txn.fetchall()
|
||||||
|
|
||||||
|
sql = (
|
||||||
|
"SELECT -event_stream_ordering, event_id, state_group"
|
||||||
|
" FROM ex_outlier_stream"
|
||||||
|
" WHERE ? > event_stream_ordering"
|
||||||
|
" AND event_stream_ordering >= ?"
|
||||||
|
" ORDER BY event_stream_ordering DESC"
|
||||||
|
)
|
||||||
|
txn.execute(sql, (last_forward_id, upper_bound))
|
||||||
|
forward_ex_outliers = txn.fetchall()
|
||||||
else:
|
else:
|
||||||
new_forward_events = []
|
new_forward_events = []
|
||||||
|
state_resets = []
|
||||||
|
forward_ex_outliers = []
|
||||||
|
|
||||||
sql = (
|
sql = (
|
||||||
"SELECT -e.stream_ordering, ej.internal_metadata, ej.json,"
|
"SELECT -e.stream_ordering, ej.internal_metadata, ej.json,"
|
||||||
@ -1128,8 +1175,35 @@ class EventsStore(SQLBaseStore):
|
|||||||
if last_backfill_id != current_backfill_id:
|
if last_backfill_id != current_backfill_id:
|
||||||
txn.execute(sql, (-last_backfill_id, -current_backfill_id, limit))
|
txn.execute(sql, (-last_backfill_id, -current_backfill_id, limit))
|
||||||
new_backfill_events = txn.fetchall()
|
new_backfill_events = txn.fetchall()
|
||||||
|
|
||||||
|
if len(new_backfill_events) == limit:
|
||||||
|
upper_bound = new_backfill_events[-1][0]
|
||||||
|
else:
|
||||||
|
upper_bound = current_backfill_id
|
||||||
|
|
||||||
|
sql = (
|
||||||
|
"SELECT -event_stream_ordering, event_id, state_group"
|
||||||
|
" FROM ex_outlier_stream"
|
||||||
|
" WHERE ? > event_stream_ordering"
|
||||||
|
" AND event_stream_ordering >= ?"
|
||||||
|
" ORDER BY event_stream_ordering DESC"
|
||||||
|
)
|
||||||
|
txn.execute(sql, (-last_backfill_id, -upper_bound))
|
||||||
|
backward_ex_outliers = txn.fetchall()
|
||||||
else:
|
else:
|
||||||
new_backfill_events = []
|
new_backfill_events = []
|
||||||
|
backward_ex_outliers = []
|
||||||
|
|
||||||
return (new_forward_events, new_backfill_events)
|
return AllNewEventsResult(
|
||||||
|
new_forward_events, new_backfill_events,
|
||||||
|
forward_ex_outliers, backward_ex_outliers,
|
||||||
|
state_resets,
|
||||||
|
)
|
||||||
return self.runInteraction("get_all_new_events", get_all_new_events_txn)
|
return self.runInteraction("get_all_new_events", get_all_new_events_txn)
|
||||||
|
|
||||||
|
|
||||||
|
AllNewEventsResult = namedtuple("AllNewEventsResult", [
|
||||||
|
"new_forward_events", "new_backfill_events",
|
||||||
|
"forward_ex_outliers", "backward_ex_outliers",
|
||||||
|
"state_resets"
|
||||||
|
])
|
||||||
|
38
synapse/storage/schema/delta/30/state_stream.sql
Normal file
38
synapse/storage/schema/delta/30/state_stream.sql
Normal file
@ -0,0 +1,38 @@
|
|||||||
|
/* Copyright 2016 OpenMarket Ltd
|
||||||
|
*
|
||||||
|
* 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.
|
||||||
|
*/
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The positions in the event stream_ordering when the current_state was
|
||||||
|
* replaced by the state at the event.
|
||||||
|
*/
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS current_state_resets(
|
||||||
|
event_stream_ordering BIGINT PRIMARY KEY NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
/* The outlier events that have aquired a state group typically through
|
||||||
|
* backfill. This is tracked separately to the events table, as assigning a
|
||||||
|
* state group change the position of the existing event in the stream
|
||||||
|
* ordering.
|
||||||
|
* However since a stream_ordering is assigned in persist_event for the
|
||||||
|
* (event, state) pair, we can use that stream_ordering to identify when
|
||||||
|
* the new state was assigned for the event.
|
||||||
|
*/
|
||||||
|
CREATE TABLE IF NOT EXISTS ex_outlier_stream(
|
||||||
|
event_stream_ordering BIGINT PRIMARY KEY NOT NULL,
|
||||||
|
event_id TEXT NOT NULL,
|
||||||
|
state_group BIGINT NOT NULL
|
||||||
|
);
|
Loading…
Reference in New Issue
Block a user