Merge pull request #5030 from matrix-org/rav/rewrite_g_s_v_k

Rewrite Datastore.get_server_verify_keys
This commit is contained in:
Richard van der Hoff 2019-04-09 11:30:23 +01:00 committed by GitHub
commit 644b86677f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 139 additions and 94 deletions

1
changelog.d/5030.misc Normal file
View File

@ -0,0 +1 @@
Rewrite Datastore.get_server_verify_keys to reduce the number of database transactions.

View File

@ -275,10 +275,6 @@ class Keyring(object):
@defer.inlineCallbacks @defer.inlineCallbacks
def do_iterations(): def do_iterations():
with Measure(self.clock, "get_server_verify_keys"): with Measure(self.clock, "get_server_verify_keys"):
# dict[str, dict[str, VerifyKey]]: results so far.
# map server_name -> key_id -> VerifyKey
merged_results = {}
# dict[str, set(str)]: keys to fetch for each server # dict[str, set(str)]: keys to fetch for each server
missing_keys = {} missing_keys = {}
for verify_request in verify_requests: for verify_request in verify_requests:
@ -288,29 +284,29 @@ class Keyring(object):
for fn in key_fetch_fns: for fn in key_fetch_fns:
results = yield fn(missing_keys.items()) results = yield fn(missing_keys.items())
merged_results.update(results)
# We now need to figure out which verify requests we have keys # We now need to figure out which verify requests we have keys
# for and which we don't # for and which we don't
missing_keys = {} missing_keys = {}
requests_missing_keys = [] requests_missing_keys = []
for verify_request in verify_requests: for verify_request in verify_requests:
server_name = verify_request.server_name
result_keys = merged_results[server_name]
if verify_request.deferred.called: if verify_request.deferred.called:
# We've already called this deferred, which probably # We've already called this deferred, which probably
# means that we've already found a key for it. # means that we've already found a key for it.
continue continue
server_name = verify_request.server_name
# see if any of the keys we got this time are sufficient to
# complete this VerifyKeyRequest.
result_keys = results.get(server_name, {})
for key_id in verify_request.key_ids: for key_id in verify_request.key_ids:
if key_id in result_keys: key = result_keys.get(key_id)
if key:
with PreserveLoggingContext(): with PreserveLoggingContext():
verify_request.deferred.callback(( verify_request.deferred.callback(
server_name, (server_name, key_id, key)
key_id, )
result_keys[key_id],
))
break break
else: else:
# The else block is only reached if the loop above # The else block is only reached if the loop above
@ -344,27 +340,24 @@ class Keyring(object):
@defer.inlineCallbacks @defer.inlineCallbacks
def get_keys_from_store(self, server_name_and_key_ids): def get_keys_from_store(self, server_name_and_key_ids):
""" """
Args: Args:
server_name_and_key_ids (list[(str, iterable[str])]): server_name_and_key_ids (iterable(Tuple[str, iterable[str]]):
list of (server_name, iterable[key_id]) tuples to fetch keys for list of (server_name, iterable[key_id]) tuples to fetch keys for
Returns: Returns:
Deferred: resolves to dict[str, dict[str, VerifyKey]]: map from Deferred: resolves to dict[str, dict[str, VerifyKey|None]]: map from
server_name -> key_id -> VerifyKey server_name -> key_id -> VerifyKey
""" """
res = yield logcontext.make_deferred_yieldable(defer.gatherResults( keys_to_fetch = (
[ (server_name, key_id)
run_in_background( for server_name, key_ids in server_name_and_key_ids
self.store.get_server_verify_keys, for key_id in key_ids
server_name, key_ids, )
).addCallback(lambda ks, server: (server, ks), server_name) res = yield self.store.get_server_verify_keys(keys_to_fetch)
for server_name, key_ids in server_name_and_key_ids keys = {}
], for (server_name, key_id), key in res.items():
consumeErrors=True, keys.setdefault(server_name, {})[key_id] = key
).addErrback(unwrapFirstError)) defer.returnValue(keys)
defer.returnValue(dict(res))
@defer.inlineCallbacks @defer.inlineCallbacks
def get_keys_from_perspectives(self, server_name_and_key_ids): def get_keys_from_perspectives(self, server_name_and_key_ids):

View File

@ -13,19 +13,9 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from synapse.storage import DataStore from synapse.storage import KeyStore
from synapse.storage.keys import KeyStore
from ._base import BaseSlavedStore, __func__ # KeyStore isn't really safe to use from a worker, but for now we do so and hope that
# the races it creates aren't too bad.
SlavedKeyStore = KeyStore
class SlavedKeyStore(BaseSlavedStore):
_get_server_verify_key = KeyStore.__dict__[
"_get_server_verify_key"
]
get_server_verify_keys = __func__(DataStore.get_server_verify_keys)
store_server_verify_key = __func__(DataStore.store_server_verify_key)
get_server_keys_json = __func__(DataStore.get_server_keys_json)
store_server_keys_json = __func__(DataStore.store_server_keys_json)

View File

@ -1,5 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd # Copyright 2014-2016 OpenMarket Ltd
# Copyright 2019 New Vector Ltd.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); # Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License. # you may not use this file except in compliance with the License.
@ -13,15 +14,15 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import itertools
import logging import logging
import six import six
from signedjson.key import decode_verify_key_bytes from signedjson.key import decode_verify_key_bytes
from twisted.internet import defer from synapse.util import batch_iter
from synapse.util.caches.descriptors import cached, cachedList
from synapse.util.caches.descriptors import cachedInlineCallbacks
from ._base import SQLBaseStore from ._base import SQLBaseStore
@ -38,36 +39,50 @@ else:
class KeyStore(SQLBaseStore): class KeyStore(SQLBaseStore):
"""Persistence for signature verification keys """Persistence for signature verification keys
""" """
@cachedInlineCallbacks()
def _get_server_verify_key(self, server_name, key_id):
verify_key_bytes = yield self._simple_select_one_onecol(
table="server_signature_keys",
keyvalues={"server_name": server_name, "key_id": key_id},
retcol="verify_key",
desc="_get_server_verify_key",
allow_none=True,
)
if verify_key_bytes: @cached()
defer.returnValue(decode_verify_key_bytes(key_id, bytes(verify_key_bytes))) def _get_server_verify_key(self, server_name_and_key_id):
raise NotImplementedError()
@defer.inlineCallbacks @cachedList(
def get_server_verify_keys(self, server_name, key_ids): cached_method_name="_get_server_verify_key", list_name="server_name_and_key_ids"
"""Retrieve the NACL verification key for a given server for the given )
key_ids def get_server_verify_keys(self, server_name_and_key_ids):
"""
Args: Args:
server_name (str): The name of the server. server_name_and_key_ids (iterable[Tuple[str, str]]):
key_ids (iterable[str]): key_ids to try and look up. iterable of (server_name, key-id) tuples to fetch keys for
Returns: Returns:
Deferred: resolves to dict[str, VerifyKey]: map from Deferred: resolves to dict[Tuple[str, str], VerifyKey|None]:
key_id to verification key. map from (server_name, key_id) -> VerifyKey, or None if the key is
unknown
""" """
keys = {} keys = {}
for key_id in key_ids:
key = yield self._get_server_verify_key(server_name, key_id) def _get_keys(txn, batch):
if key: """Processes a batch of keys to fetch, and adds the result to `keys`."""
keys[key_id] = key
defer.returnValue(keys) # batch_iter always returns tuples so it's safe to do len(batch)
sql = (
"SELECT server_name, key_id, verify_key FROM server_signature_keys "
"WHERE 1=0"
) + " OR (server_name=? AND key_id=?)" * len(batch)
txn.execute(sql, tuple(itertools.chain.from_iterable(batch)))
for row in txn:
server_name, key_id, key_bytes = row
keys[(server_name, key_id)] = decode_verify_key_bytes(
key_id, bytes(key_bytes)
)
def _txn(txn):
for batch in batch_iter(server_name_and_key_ids, 50):
_get_keys(txn, batch)
return keys
return self.runInteraction("get_server_verify_keys", _txn)
def store_server_verify_key( def store_server_verify_key(
self, server_name, from_server, time_now_ms, verify_key self, server_name, from_server, time_now_ms, verify_key
@ -93,8 +108,11 @@ class KeyStore(SQLBaseStore):
"verify_key": db_binary_type(verify_key.encode()), "verify_key": db_binary_type(verify_key.encode()),
}, },
) )
# invalidate takes a tuple corresponding to the params of
# _get_server_verify_key. _get_server_verify_key only takes one
# param, which is itself the 2-tuple (server_name, key_id).
txn.call_after( txn.call_after(
self._get_server_verify_key.invalidate, (server_name, key_id) self._get_server_verify_key.invalidate, ((server_name, key_id),)
) )
return self.runInteraction("store_server_verify_key", _txn) return self.runInteraction("store_server_verify_key", _txn)

View File

@ -15,34 +15,77 @@
import signedjson.key import signedjson.key
from twisted.internet import defer from twisted.internet.defer import Deferred
import tests.unittest import tests.unittest
import tests.utils
KEY_1 = signedjson.key.decode_verify_key_base64(
"ed25519", "key1", "fP5l4JzpZPq/zdbBg5xx6lQGAAOM9/3w94cqiJ5jPrw"
)
KEY_2 = signedjson.key.decode_verify_key_base64(
"ed25519", "key2", "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw"
)
class KeyStoreTestCase(tests.unittest.TestCase): class KeyStoreTestCase(tests.unittest.HomeserverTestCase):
@defer.inlineCallbacks
def setUp(self):
hs = yield tests.utils.setup_test_homeserver(self.addCleanup)
self.store = hs.get_datastore()
@defer.inlineCallbacks
def test_get_server_verify_keys(self): def test_get_server_verify_keys(self):
key1 = signedjson.key.decode_verify_key_base64( store = self.hs.get_datastore()
"ed25519", "key1", "fP5l4JzpZPq/zdbBg5xx6lQGAAOM9/3w94cqiJ5jPrw"
)
key2 = signedjson.key.decode_verify_key_base64(
"ed25519", "key2", "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw"
)
yield self.store.store_server_verify_key("server1", "from_server", 0, key1)
yield self.store.store_server_verify_key("server1", "from_server", 0, key2)
res = yield self.store.get_server_verify_keys( d = store.store_server_verify_key("server1", "from_server", 0, KEY_1)
"server1", ["ed25519:key1", "ed25519:key2", "ed25519:key3"] self.get_success(d)
) d = store.store_server_verify_key("server1", "from_server", 0, KEY_2)
self.get_success(d)
d = store.get_server_verify_keys(
[
("server1", "ed25519:key1"),
("server1", "ed25519:key2"),
("server1", "ed25519:key3"),
]
)
res = self.get_success(d)
self.assertEqual(len(res.keys()), 3)
self.assertEqual(res[("server1", "ed25519:key1")].version, "key1")
self.assertEqual(res[("server1", "ed25519:key2")].version, "key2")
# non-existent result gives None
self.assertIsNone(res[("server1", "ed25519:key3")])
def test_cache(self):
"""Check that updates correctly invalidate the cache."""
store = self.hs.get_datastore()
key_id_1 = "ed25519:key1"
key_id_2 = "ed25519:key2"
d = store.store_server_verify_key("srv1", "from_server", 0, KEY_1)
self.get_success(d)
d = store.store_server_verify_key("srv1", "from_server", 0, KEY_2)
self.get_success(d)
d = store.get_server_verify_keys([("srv1", key_id_1), ("srv1", key_id_2)])
res = self.get_success(d)
self.assertEqual(len(res.keys()), 2) self.assertEqual(len(res.keys()), 2)
self.assertEqual(res["ed25519:key1"].version, "key1") self.assertEqual(res[("srv1", key_id_1)], KEY_1)
self.assertEqual(res["ed25519:key2"].version, "key2") self.assertEqual(res[("srv1", key_id_2)], KEY_2)
# we should be able to look up the same thing again without a db hit
res = store.get_server_verify_keys([("srv1", key_id_1)])
if isinstance(res, Deferred):
res = self.successResultOf(res)
self.assertEqual(len(res.keys()), 1)
self.assertEqual(res[("srv1", key_id_1)], KEY_1)
new_key_2 = signedjson.key.get_verify_key(
signedjson.key.generate_signing_key("key2")
)
d = store.store_server_verify_key("srv1", "from_server", 10, new_key_2)
self.get_success(d)
d = store.get_server_verify_keys([("srv1", key_id_1), ("srv1", key_id_2)])
res = self.get_success(d)
self.assertEqual(len(res.keys()), 2)
self.assertEqual(res[("srv1", key_id_1)], KEY_1)
self.assertEqual(res[("srv1", key_id_2)], new_key_2)