Move catchup of replication streams to worker.

[erikjohnston]

This changes the replication protocol so that the server does not send down RDATA for rows that happened before the client connected. Instead, the server will send a POSITION and clients then query the database (or master out of band) to get up to date.

Based on #7010 and #7011

[erikjohnston]

(Test failures are due to sytest being broken w.r.t. CAS login)

[richvdh]

this is too big for me to keep in my head, so I can't really check it for correctness; I'll have to take your word for it.

It seems generally sane, but a few thoughts below.

[richvdh]

where is this used?

[erikjohnston]

This gets used by the TypingStream.current_token, which now gets called on workers (this is the equivalent of calling a slaved ID gen current_token function).

[richvdh]

I can't help but feel that this would be much easier to follow if it were implemented in the individual Streams:

if hs.config.worker_app is None:
    self.update_function = handler.get_updates
else:
    self.update_function = <something to do with ReplicationGetStreamUpdates>

[erikjohnston]

This turns out to be surprisingly faffy, given that what the API call returns is not quite the same as what the update_function returns (as can be seen from the differences in the two branches below the commented line). I guess we could make them be identical (i.e. make the API hit only return the rows), but I'm not sure that really makes things clearer. Thoughts?

[richvdh]

Make the spec for update_function be that it return the same (updates, new_last_token, limited) tuple.

Have a helper function:

def db_query_to_update_function(query_function):
    """Wraps a db query function which returns a list of rows to make it suitable for use as an `update_function` for the Stream class"""
    async def update_function(from_token, upto_token, limit):
            rows = await query_function(from_token, upto_token, limit=limit)
            updates = [(row[0], row[1:]) for row in rows]
            limited = False
            if len(updates) == limit:
                upto_token = rows[-1][0]
                limited = True

            return updates, upto_token, limited
    return update_function

Do something similar with the ReplicationClient.

Use that to wrap the db query function in the Stream class:

        self.update_function = db_query_to_update_function(store.get_all_new_backfill_event_rows)

Suggest sorting out the type annotations to help with that.

I think it will be clearer than adding more and more fields to the Stream class to control how it is supposed to work.

[erikjohnston]

Ok, let me give that a go. (Annoyingly mypy refuses to accept setting functions like we do in stream, hence the type: ignore going on, which is apain)

[erikjohnston]

Oh, I messed that up.

[richvdh]

presumably if update_function wasn't defined as a method in the base class, the type:ignore wouldn't be necessary?

[erikjohnston]

Well, then it complains that Stream doesn't have a update_function, so you add one as a property and then it complains when you try and declare async def update_function(..) in sub classes.

[richvdh]

feels like we should make some type definitions somewhere...

[erikjohnston]

Yeeeeeeeeeeeeeees.

[erikjohnston]

Ok, I've added some type aliases, although it didn't really simplify things very much in the end :/

[richvdh]

well I was really thinking we should have something like:

UpdateFunctionResult = Tuple[List[StreamRow], Token, bool]
UpdateFunction = Callable[[Token, Token, int], Awaitable[UpdateFunctionResult]]

but shrugface

[richvdh]

lgtm

[erikjohnston]

Thanking you! ❤️