Replication fails with AssertionError in replication_notifier

[richvdh]

my master stopped sending event notifications over replication. It was logging this sort of thing:

2020-04-20 09:36:14,977 - synapse.replication.tcp.resource - 169 - INFO - replication_notifier-231759- Failed to handle stream events
2020-04-20 09:36:14,984 - synapse.metrics.background_process_metrics - 215 - ERROR - replication_notifier-231759- Background process 'replication_notifier' threw an exception
Capture point (most recent call last):
... lots of output...
During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/synapse/synapse/synapse/metrics/background_process_metrics.py", line 213, in run
    return (yield result)
  File "/opt/synapse/env3/lib/python3.5/site-packages/twisted/internet/defer.py", line 1418, in _inlineCallbacks
    result = g.send(result)
  File "/opt/synapse/synapse/synapse/replication/tcp/resource.py", line 166, in _run_notifier_loop
    updates, current_token, limited = await stream.get_updates()
  File "/opt/synapse/synapse/synapse/replication/tcp/streams/_base.py", line 134, in get_updates
    self.last_token, current_token
  File "/opt/synapse/synapse/synapse/replication/tcp/streams/_base.py", line 159, in get_updates_since
    from_token, upto_token, limit,
  File "/opt/synapse/synapse/synapse/replication/tcp/streams/_base.py", line 178, in update_function
    assert len(updates) <= limit
AssertionError

Likely this only affects develop currently.

[richvdh]

#7337 has fixed the primary cause of this, but there's more work to be done:

• we should test the new code properly
• it still won't work properly if we hit the limit on the current_state_delta stream
• we should check that other streams aren't affected (though that is More AssertionErrors from replication_notifier #7340)

[richvdh]

turns out there are more of these:

Traceback (most recent call last):
  File "/opt/synapse/synapse/synapse/metrics/background_process_metrics.py", line 213, in run
    return (yield result)
  File "/opt/synapse/env3/lib/python3.5/site-packages/twisted/internet/defer.py", line 1418, in _inlineCallbacks
    result = g.send(result)
  File "/opt/synapse/synapse/synapse/replication/tcp/resource.py", line 166, in _run_notifier_loop
    updates, current_token, limited = await stream.get_updates()
  File "/opt/synapse/synapse/synapse/replication/tcp/streams/_base.py", line 138, in get_updates
    self.last_token, current_token
  File "/opt/synapse/synapse/synapse/replication/tcp/streams/_base.py", line 163, in get_updates_since
    from_token, upto_token, _STREAM_UPDATE_TARGET_ROW_COUNT,
  File "/opt/synapse/synapse/synapse/replication/tcp/streams/events.py", line 180, in _update_function
    assert len(state_rows) < target_row_count
AssertionError

This looks like it is essentially:

it still won't work properly if we hit the limit on the current_state_delta stream

[richvdh]

File "/opt/synapse/synapse/synapse/replication/tcp/streams/events.py", line 180, in _update_function
assert len(state_rows) < target_row_count

Hopefully this is fixed for reals by #7358.