Merge pull request #781 from matrix-org/markjh/replication_problems
Fix a bug in replication that was causing the pusher to tight looppull/783/head
commit
cdda850ce1
|
@ -164,8 +164,8 @@ class ReplicationResource(Resource):
|
||||||
"Replicating %d rows of %s from %s -> %s",
|
"Replicating %d rows of %s from %s -> %s",
|
||||||
len(stream_content["rows"]),
|
len(stream_content["rows"]),
|
||||||
stream_name,
|
stream_name,
|
||||||
stream_content["position"],
|
|
||||||
request_streams.get(stream_name),
|
request_streams.get(stream_name),
|
||||||
|
stream_content["position"],
|
||||||
)
|
)
|
||||||
|
|
||||||
request.write(json.dumps(result, ensure_ascii=False))
|
request.write(json.dumps(result, ensure_ascii=False))
|
||||||
|
|
|
@ -146,12 +146,14 @@ class SlavedEventStore(BaseSlavedStore):
|
||||||
|
|
||||||
stream = result.get("forward_ex_outliers")
|
stream = result.get("forward_ex_outliers")
|
||||||
if stream:
|
if stream:
|
||||||
|
self._stream_id_gen.advance(stream["position"])
|
||||||
for row in stream["rows"]:
|
for row in stream["rows"]:
|
||||||
event_id = row[1]
|
event_id = row[1]
|
||||||
self._invalidate_get_event_cache(event_id)
|
self._invalidate_get_event_cache(event_id)
|
||||||
|
|
||||||
stream = result.get("backward_ex_outliers")
|
stream = result.get("backward_ex_outliers")
|
||||||
if stream:
|
if stream:
|
||||||
|
self._backfill_id_gen.advance(-stream["position"])
|
||||||
for row in stream["rows"]:
|
for row in stream["rows"]:
|
||||||
event_id = row[1]
|
event_id = row[1]
|
||||||
self._invalidate_get_event_cache(event_id)
|
self._invalidate_get_event_cache(event_id)
|
||||||
|
|
Loading…
Reference in New Issue