Add endpoints for backfilling history (MSC2716)

[MadLittleMods]

Implement MSC2716 to add endpoints for backfilling history. This PR does not support federation use cases with the "marker" and "insertion" events.

For reviewers, it's probably best to see this in action with the associated Complement tests.

---

Complement MR: matrix-org/complement#68

TARDIS visualization MR: matrix-org/tardis#1

Getting started

The PR adds the POST /_matrix/client/unstable/org.matrix.msc2716/rooms/<roomID>/batch_send?prev_event=<eventID>&chunk_id=<chunkID> endpoint which can insert a chunk of events historically back in time next to the given prev_event. chunk_id comes from next_chunk_id in the response of the batch send endpoint and is derived from the "insertion" events added to each chunk. It's not required for the first batch send.

{
    "events": [ ... ],
    "state_events_at_start": [ ... ]
}

The /batchsend endpoint is behind a feature flag: experimental_features -> msc2716_enabled (defined in homseserver.yaml). And is only available to application services so you will need to add one to your homeserver.yml and use the as_token defined to interact with the API (other tokens will 403).

state_events_at_start is used to define the historical state events needed to auth the events like join events. These events will float outside of the normal DAG as outlier's and won't be visible in the chat history which also allows us to insert multiple chunks without having a bunch of @mxid joined the room noise between each chunk.

events is chronological(oldest to newest) chunk/list of events you want to insert. There is a reverse-chronological constraint on chunks so once you insert some messages, you can only insert older ones after that. tldr; Insert from your most recent history -> oldest history.

Why? depth is not re-calculated when historical messages are inserted into the DAG. This means we have to take care to insert in the right order. Events are sorted by (topological_ordering, stream_ordering) where topological_ordering is just depth. Normally, stream_ordering is an auto incrementing integer but for backfilled=true events, it decrements. Historical messages are inserted all at the same depth, and marked as backfilled so the stream_ordering decrements and each event is sorted behind the next. (from #9247 (comment))

If you're curious to look at a known working example, the Complement tests have barebones test cases interacting with this API, matrix-org/complement#68

Steps to reproduce:

1. In your homeserver.yaml, add the feature flag to enable the /batchsend endpoint

   experimental_features:
     # Enable history backfilling support
     msc2716_enabled: true

2. Define an application service in your homeserver.yaml. This could be one of your existing bridges. See the application service guide for an example of what the registration file would look like. We only care about the as_token in this case.

   app_service_config_files:
     - /data/my-as-registration.yaml

3. POST /_matrix/client/unstable/org.matrix.msc2716/rooms/<roomID>/batch_send?prev_event=<eventID>&chunk_id=<chunkID> with the Authorization: Bearer <as_token> header and the following body:
   - prev_event is the event you want to insert next to. Your historical messages will appear after this event so pick one where the timestamp makes sense. To be a little more idiomatic for inserting historical events that happened before the Matrix room creation, prev_event could be some primordial creation event for the room.
   - chunk_id comes from next_chunk_id in the previous batch send response. It connects the last (most recent) message to the insertion event of the previous chunk. The parameter is not needed for your first chunk because there is nothing to connect to yet. Note: The messages will appear correctly on your local server without it but it's important to have this set for federated servers so messages backfill correctly.
   - You can change origin_server_ts in your events to whatever you want to display it as
   - Add a join event in state_events_at_start for any message author in the events
   - m.historical will automatically be added to each your events. This is important to mark them as backfilled, sort correctly, and skip the push notifications actions.

   {
       "events": [{
           "type": "m.room.message",
           "sender": "@maria:hs1",
           "origin_server_ts": 1620336731128,
           "content": {
               "msgtype": "m.text",
               "body": "Some historical message",
               "m.historical": true,
           },
       }],
       "state_events_at_start": [{
           "type": "m.room.member",
           "sender": "@maria:hs1",
           "origin_server_ts": 1620336731128,
           "content": {
               "membership": "join",
           },
           "state_key": "@maria:hs1",
       }]
   }

Dev notes

COMPLEMENT_DIR=../complement ./scripts-dev/complement.sh

./scripts-dev/complement.sh

---

API:

• PUT /_matrix/client/r0/rooms/{roomId}/send/{eventType}/{txnId}
• GET /_matrix/client/r0/rooms/{roomId}/messages
• https://matrix.org/docs/spec/rooms/v4#persistent-data-unit-schema
• https://matrix.org/docs/spec/client_server/r0.6.1#room-events

Relevant code:

• scripts-dev/complement.sh
• synapse/rest/client/v1/room.py#L218 -> /rooms/$roomid/send/$event_type[/$txn_id]] PATTERNS = "/rooms/(?P<room_id>[^/]*)/send/(?P<event_type>[^/]*)"
• synapse/handlers/message.py#L683 -> event_creation_handler.create_and_send_nonmember_event
• I noticed in the Synapse code that we already accept a ts query parameter to override the origin_server_ts if the request is coming from an app service, synapse/rest/client/v1/room.py#L233-L234 (via add ?ts massaging for ASes #2754)
• synapse/federation/transport/server.py#L372

event_creation_handler.create_and_send_nonmember_event
EventCreationHandler.create_event
EventCreationHandler.create_new_client_event
EventBuilder.build

_generate_local_out_of_band_leave

---

persist_events or persist_event
_maybe_start_persisting
_persist_events
_persist_events_and_state_updates
_persist_events_txn
_update_metadata_tables_txn
_handle_mult_prev_events
-> simple_insert_many_txn `event_edges`

Call stacks for passing depth, prev_event_ids, auth_event_ids:

create_and_send_nonmember_event
create_event
create_new_client_event
builder.build

update_membership
update_membership_locked
_local_membership_update
create_event

---

COMPLEMENT_BASE_IMAGE=complement-synapse go test -tags msc2716 -v -count=1 ./tests/main_test.go ./tests/msc2716_test.go

---

• https://matrix.org/docs/spec/server_server/unstable#put-matrix-federation-v1-send-txnid

Accessing the database

Access the sqlite database in the Docker container after Complement runs. Be sure to change the defer deployment.Destroy(t) call in the Complement tests to a defer time.Sleep(2 * time.Hour) so the Docker container stays alive after it's finished.

$ docker exec -it b91db4912057 /bin/bash
$ apt-get update
$ apt-get install sqlite3
$ sqlite3 /conf/homeserver.db

.tables
.schema events

select * from events where event_id='$LfK8zWi0g_snvqwi93vWGKj-iTf7gxWh_WRhTD5pALc';
select * from event_json where event_id='$LfK8zWi0g_snvqwi93vWGKj-iTf7gxWh_WRhTD5pALc';
select * from event_auth where event_id='$LfK8zWi0g_snvqwi93vWGKj-iTf7gxWh_WRhTD5pALc';
select * from event_auth_chains where event_id='$LfK8zWi0g_snvqwi93vWGKj-iTf7gxWh_WRhTD5pALc';
select * from event_auth_chain_to_calculate where event_id='$LfK8zWi0g_snvqwi93vWGKj-iTf7gxWh_WRhTD5pALc';

Event signing

• https://matrix.org/docs/spec/server_server/r0.1.4#id14
  • Old docs: https://matrix.org/docs/spec/r0.0.1/server_server#signing-events
• https://spec.matrix.org/unstable/server-server-api/#signing-events

test-event-signing.py (v1)

test-event-signing.py

from signedjson.key import (
    NACL_ED25519,
    decode_signing_key_base64,
    decode_verify_key_bytes,
    decode_verify_key_base64,
    generate_signing_key,
    get_verify_key,
    is_signing_algorithm_supported,
    read_signing_keys,
    write_signing_keys,
)
from signedjson.sign import (
    sign_json, verify_signed_json, SignatureVerifyException
)

SERVER_NAME = 'hs1'
KEY_ID = 'ed25519:a_imNW';
# This from the signature key_id `ed25519:a_FBvY`
VERSION = 'a_imNW'
SIGNING_KEY_BASE_64 = "vX2BK8l89Qk2EAVcaCiGiTUWG59dwleotTjzqu80C4w"
VERIFY_KEY_BASE_64 = "9d92WUgYwsKY0oWxOR1R61SJar9+D7uvz59IEDEjqyI"
JSON = {'my_key': 'my_data'}
# JSON = {
# 	'auth_events': ['$M7y6Yy1agkTx4eLXpZRndF1BdGk5pvCkjNCZkKlGIdY', '$ZtnAy9W445OtyeqCEk9Dcn6Nr14kYb4z-fJM8MwcVmI', '$gf4M_E3lR3QhumzTKMCncsDrTLZSVjuhcHZ4y6EtJzA', '$mSfCznaiy9xZvR7uQb7NdbNVS-1DNSe7lilB649H78g', '$8WTDdQw82NY8SUWUom53IjM1HzXiaB71oeU1mPjz7XY', '$ywbCJgiKeXtXuhN8OzxFD04-Esj89nxSt5miRxM5f78', '$ooT281Jg3nfzhjdE1rq0hUEOpu8sX1ublFUDYRPuyOk'], 'prev_events': ['$zWKxzzgcSVoLUDAeauzbOXkCTfrfNCcRQwEQToTZpof'],
# 	'type': 'm.room.member',
# 	'room_id': '!aqtKBNDKZSWCDMRpSH:hs1',
# 	'sender': '@maria:hs1',
# 	'content': {'membership': 'join'},
# 	'depth': 1,
# 	'prev_state': [],
# 	'state_key':
# 	'@maria:hs1',
# 	'origin': 'hs1', 
# 	'origin_server_ts': 1620167174120,
# 	'hashes': {'sha256': 'Cw+4O3jZVocrTLhdJ9vfoZq1SRYk4RB+/8X5EQ/xaXU'},
# 	'signatures': {'hs1': {'ed25519:a_imNW': '8/II10y7FO7nMBZLeF6aRMkDAkSjnRsJIu8oPGfG9AEvmvEvusNwe6klWfw7QPB0NfJpF5wyOd5rQrbX7ymcAg'}},
# 	'unsigned': {}
# }


#signing_key = generate_signing_key('zxcvb')
#verify_key = get_verify_key(signing_key)

signing_key = decode_signing_key_base64(NACL_ED25519, VERSION, SIGNING_KEY_BASE_64)
#verify_key = decode_verify_key_bytes(KEY_ID, VERIFY_KEY_BASE_64)
verify_key = decode_verify_key_base64(NACL_ED25519, VERSION, VERIFY_KEY_BASE_64)
verify_key_derived_from_signing_key = get_verify_key(signing_key)
if verify_key != verify_key_derived_from_signing_key:
	print("WARNING: verify_key and verify_key_derived_from_signing_key are different")

print(f"signing_key={signing_key}")
print(f"verify_key={verify_key} verify_key_derived_from_signing_key={verify_key_derived_from_signing_key}")

signed_json = sign_json(JSON, SERVER_NAME, signing_key)


try:
    verify_signed_json(signed_json, SERVER_NAME, verify_key)
    print('Signature is valid')
except SignatureVerifyException:
    print('Signature is invalid')

test-event-signing.py (v2)

test-event-signing.py

import copy
from signedjson.key import (
    NACL_ED25519,
    decode_signing_key_base64,
    decode_verify_key_bytes,
    decode_verify_key_base64,
    generate_signing_key,
    get_verify_key,
    is_signing_algorithm_supported,
    read_signing_keys,
    write_signing_keys,
)
from signedjson.sign import (
    sign_json, verify_signed_json, SignatureVerifyException
)

SERVER_NAME = 'hs1'
KEY_ID = 'ed25519:a_imNW';
# This from the signature key_id `ed25519:a_FBvY`
VERSION = 'a_imNW'
SIGNING_KEY_BASE_64 = "vX2BK8l89Qk2EAVcaCiGiTUWG59dwleotTjzqu80C4w"
VERIFY_KEY_BASE_64 = "9d92WUgYwsKY0oWxOR1R61SJar9+D7uvz59IEDEjqyI"


#signing_key = generate_signing_key('zxcvb')
#verify_key = get_verify_key(signing_key)

signing_key = decode_signing_key_base64(NACL_ED25519, VERSION, SIGNING_KEY_BASE_64)
#verify_key = decode_verify_key_bytes(KEY_ID, VERIFY_KEY_BASE_64)
verify_key = decode_verify_key_base64(NACL_ED25519, VERSION, VERIFY_KEY_BASE_64)
verify_key_derived_from_signing_key = get_verify_key(signing_key)
if verify_key != verify_key_derived_from_signing_key:
	print("WARNING: verify_key and verify_key_derived_from_signing_key are different")

print(f"signing_key={signing_key}")
print(f"verify_key={verify_key} verify_key_derived_from_signing_key={verify_key_derived_from_signing_key}")

#JSON = {'my_key': 'my_data'}
#signed_json = sign_json(JSON, SERVER_NAME, signing_key)
KNOWN_GOOD_JSON = {
	'auth_events': ['$M7y6Yy1agkTx4eLXpZRndF1BdGk5pvCkjNCZkKlGIdY'],
	'prev_events': ['$M7y6Yy1agkTx4eLXpZRndF1BdGk5pvCkjNCZkKlGIdY'],
	'type': 'm.room.member',
	'room_id': '!aqtKBNDKZSWCDMRpSH:hs1',
	'sender': '@the-bridge-user:hs1',
	'content': {'membership': 'join'},
	'depth': 2,
	'prev_state': [],
	'state_key': '@the-bridge-user:hs1',
	'origin': 'hs1',
	'origin_server_ts': 1620167173759,
	'hashes': {'sha256': 'FSyUP1X9VrlCCPztDLooSZSjcWqbUV2T7j7Z3o06zSw'},
	'signatures': {'hs1': {'ed25519:a_imNW': 'CO6WUNuoZb8bg0cH9zoywZWqzEc2YogZsp6jqhISjhIvv/HDMJYf0INlxYpo3m67xcXVYrh1LgeVw4qSRFWPDQ'}},
	'unsigned': {}
}
KNOWN_BAD_JSON = {
	'auth_events': ['$M7y6Yy1agkTx4eLXpZRndF1BdGk5pvCkjNCZkKlGIdY', '$ZtnAy9W445OtyeqCEk9Dcn6Nr14kYb4z-fJM8MwcVmI', '$gf4M_E3lR3QhumzTKMCncsDrTLZSVjuhcHZ4y6EtJzA', '$mSfCznaiy9xZvR7uQb7NdbNVS-1DNSe7lilB649H78g', '$8WTDdQw82NY8SUWUom53IjM1HzXiaB71oeU1mPjz7XY', '$ywbCJgiKeXtXuhN8OzxFD04-Esj89nxSt5miRxM5f78', '$ooT281Jg3nfzhjdE1rq0hUEOpu8sX1ublFUDYRPuyOk'],
	'prev_events': ['$zWKxzzgcSVoLUDAeauzbOXkCTfrfNCcRQwEQToTZpof'],
	'type': 'm.room.member',
	'room_id': '!aqtKBNDKZSWCDMRpSH:hs1',
	'sender': '@maria:hs1',
	'content': {'membership': 'join'},
	'depth': 1,
	'prev_state': [],
	'state_key':
	'@maria:hs1',
	'origin': 'hs1', 
	'origin_server_ts': 1620167174120,
	'hashes': {'sha256': 'Cw+4O3jZVocrTLhdJ9vfoZq1SRYk4RB+/8X5EQ/xaXU'},
	'signatures': {'hs1': {'ed25519:a_imNW': '8/II10y7FO7nMBZLeF6aRMkDAkSjnRsJIu8oPGfG9AEvmvEvusNwe6klWfw7QPB0NfJpF5wyOd5rQrbX7ymcAg'}},
	'unsigned': {}
}
# Switch around KNOWN_GOOD_JSON and KNOWN_BAD_JSON here
signed_json = KNOWN_BAD_JSON


copied_json = copy.deepcopy(signed_json)
double_checked_signed_json = sign_json(copied_json, SERVER_NAME, signing_key)
double_checked_signatures = double_checked_signed_json.get("signatures", {})
double_checked_server_signature = double_checked_signatures.get(SERVER_NAME, {}).get(KEY_ID)
print(f"double_check_signed_json={double_checked_signatures}")
server_signature = signed_json.get("signatures", {}).get(SERVER_NAME, {}).get(KEY_ID)
if double_checked_server_signature != server_signature:
	print(f"WANRING: When we re-signed the object and checked the signatures, they did NOT match!\ndouble_checked_server_signature={double_checked_server_signature}\nserver_signature={server_signature}")


try:
    verify_signed_json(signed_json, SERVER_NAME, verify_key)
    print('Signature is valid')
except SignatureVerifyException:
    print('Signature is invalid')

Todo

• ts query param to override origin_server_ts
• prev_event query param
• Proper depth
• m.historical event field
• Add tests within Synapse

Pull Request Checklist

• Pull request is based on the develop branch
• Pull request includes a changelog file. The entry should:
  • Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from EventStore to EventWorkerStore.".
  • Use markdown where necessary, mostly for code blocks.
  • End with either a period (.) or an exclamation mark (!).
  • Start with a capital letter.
• Pull request includes a sign off
• Code style is correct (run the linters)

[clokep]

@MadLittleMods This seems to have some style / CI issues. Were you looking for general feedback or were you hoping to get this merged?

[MadLittleMods]

@clokep Some general comments and answer questions in the PR. What tests and where to add, etc?

What's more needed content wise to merge. Outside are a few lints to clean up.

Behind feature flag or unstable build type of thing?

[erikjohnston]

On the complemenet tests: you may need to merge in latest develop into this branch and the same named branch on complement repo (or delete that branch)? The tests failures look to be mostly for things like spaces and knocking, which have recently had work done on them?

[erikjohnston]

🎉

[MadLittleMods]

Woot! Thank you @erikjohnston for all the help guiding this along and review to push this to the right places! 🐗