diff --git a/synapse/api/errors.py b/synapse/api/errors.py
index 6fbd5d6876..d0dfa959dc 100644
--- a/synapse/api/errors.py
+++ b/synapse/api/errors.py
@@ -66,6 +66,17 @@ class CodeMessageException(RuntimeError):
return cs_error(self.msg)
+class MatrixCodeMessageException(CodeMessageException):
+ """An error from a general matrix endpoint, eg. from a proxied Matrix API call.
+
+ Attributes:
+ errcode (str): Matrix error code e.g 'M_FORBIDDEN'
+ """
+ def __init__(self, code, msg, errcode=Codes.UNKNOWN):
+ super(MatrixCodeMessageException, self).__init__(code, msg)
+ self.errcode = errcode
+
+
class SynapseError(CodeMessageException):
"""A base exception type for matrix errors which have an errcode and error
message (as well as an HTTP status code).
diff --git a/synapse/events/snapshot.py b/synapse/events/snapshot.py
index 6be18880b9..e9a732ff03 100644
--- a/synapse/events/snapshot.py
+++ b/synapse/events/snapshot.py
@@ -50,6 +50,7 @@ class EventContext(object):
"prev_group",
"delta_ids",
"prev_state_events",
+ "app_service",
]
def __init__(self):
@@ -68,3 +69,5 @@ class EventContext(object):
self.delta_ids = None
self.prev_state_events = None
+
+ self.app_service = None
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index ebbf844489..2af9849ed0 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -171,6 +171,16 @@ class FederationHandler(BaseHandler):
yield self._get_missing_events_for_pdu(
origin, pdu, prevs, min_depth
)
+
+ # Update the set of things we've seen after trying to
+ # fetch the missing stuff
+ have_seen = yield self.store.have_events(prevs)
+ seen = set(have_seen.iterkeys())
+
+ if not prevs - seen:
+ logger.info(
+ "Found all missing prev events for %s", pdu.event_id
+ )
elif prevs - seen:
logger.info(
"Not fetching %d missing events for room %r,event %s: %r...",
@@ -178,8 +188,6 @@ class FederationHandler(BaseHandler):
list(prevs - seen)[:5],
)
- prevs = {e_id for e_id, _ in pdu.prev_events}
- seen = set(have_seen.keys())
if prevs - seen:
logger.info(
"Still missing %d events for room %r: %r...",
@@ -214,19 +222,15 @@ class FederationHandler(BaseHandler):
Args:
origin (str): Origin of the pdu. Will be called to get the missing events
pdu: received pdu
- prevs (str[]): List of event ids which we are missing
+ prevs (set(str)): List of event ids which we are missing
min_depth (int): Minimum depth of events to return.
-
- Returns:
- Deferred<dict(str, str?)>: updated have_seen dictionary
"""
# We recalculate seen, since it may have changed.
have_seen = yield self.store.have_events(prevs)
seen = set(have_seen.keys())
if not prevs - seen:
- # nothing left to do
- defer.returnValue(have_seen)
+ return
latest = yield self.store.get_latest_event_ids_in_room(
pdu.room_id
@@ -288,19 +292,6 @@ class FederationHandler(BaseHandler):
get_missing=False
)
- have_seen = yield self.store.have_events(
- [ev for ev, _ in pdu.prev_events]
- )
- seen = set(have_seen.keys())
- if prevs - seen:
- logger.info(
- "Still missing %d prev events for %s: %r...",
- len(prevs - seen), pdu.event_id, list(prevs - seen)[:5]
- )
- else:
- logger.info("Found all missing prev events for %s", pdu.event_id)
- defer.returnValue(have_seen)
-
@log_function
@defer.inlineCallbacks
def _process_received_pdu(self, origin, pdu, state, auth_chain):
diff --git a/synapse/handlers/identity.py b/synapse/handlers/identity.py
index 6a53c5eb47..9efcdff1d6 100644
--- a/synapse/handlers/identity.py
+++ b/synapse/handlers/identity.py
@@ -18,7 +18,7 @@
from twisted.internet import defer
from synapse.api.errors import (
- CodeMessageException
+ MatrixCodeMessageException, CodeMessageException
)
from ._base import BaseHandler
from synapse.util.async import run_on_reactor
@@ -90,6 +90,9 @@ class IdentityHandler(BaseHandler):
),
{'sid': creds['sid'], 'client_secret': client_secret}
)
+ except MatrixCodeMessageException as e:
+ logger.info("getValidated3pid failed with Matrix error: %r", e)
+ raise SynapseError(e.code, e.msg, e.errcode)
except CodeMessageException as e:
data = json.loads(e.msg)
@@ -159,6 +162,9 @@ class IdentityHandler(BaseHandler):
params
)
defer.returnValue(data)
+ except MatrixCodeMessageException as e:
+ logger.info("Proxied requestToken failed with Matrix error: %r", e)
+ raise SynapseError(e.code, e.msg, e.errcode)
except CodeMessageException as e:
logger.info("Proxied requestToken failed: %r", e)
raise e
@@ -193,6 +199,9 @@ class IdentityHandler(BaseHandler):
params
)
defer.returnValue(data)
+ except MatrixCodeMessageException as e:
+ logger.info("Proxied requestToken failed with Matrix error: %r", e)
+ raise SynapseError(e.code, e.msg, e.errcode)
except CodeMessageException as e:
logger.info("Proxied requestToken failed: %r", e)
raise e
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 82a2ade1f6..57265c6d7d 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -175,7 +175,8 @@ class MessageHandler(BaseHandler):
defer.returnValue(chunk)
@defer.inlineCallbacks
- def create_event(self, event_dict, token_id=None, txn_id=None, prev_event_ids=None):
+ def create_event(self, requester, event_dict, token_id=None, txn_id=None,
+ prev_event_ids=None):
"""
Given a dict from a client, create a new event.
@@ -185,6 +186,7 @@ class MessageHandler(BaseHandler):
Adds display names to Join membership events.
Args:
+ requester
event_dict (dict): An entire event
token_id (str)
txn_id (str)
@@ -226,6 +228,7 @@ class MessageHandler(BaseHandler):
event, context = yield self._create_new_client_event(
builder=builder,
+ requester=requester,
prev_event_ids=prev_event_ids,
)
@@ -319,6 +322,7 @@ class MessageHandler(BaseHandler):
See self.create_event and self.send_nonmember_event.
"""
event, context = yield self.create_event(
+ requester,
event_dict,
token_id=requester.access_token_id,
txn_id=txn_id
@@ -416,7 +420,7 @@ class MessageHandler(BaseHandler):
@measure_func("_create_new_client_event")
@defer.inlineCallbacks
- def _create_new_client_event(self, builder, prev_event_ids=None):
+ def _create_new_client_event(self, builder, requester=None, prev_event_ids=None):
if prev_event_ids:
prev_events = yield self.store.add_event_hashes(prev_event_ids)
prev_max_depth = yield self.store.get_max_depth_of_events(prev_event_ids)
@@ -456,6 +460,8 @@ class MessageHandler(BaseHandler):
state_handler = self.state_handler
context = yield state_handler.compute_event_context(builder)
+ if requester:
+ context.app_service = requester.app_service
if builder.is_state():
builder.prev_state = yield self.store.add_event_hashes(
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index 28b2c80a93..ab87632d99 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -70,6 +70,7 @@ class RoomMemberHandler(BaseHandler):
content["kind"] = "guest"
event, context = yield msg_handler.create_event(
+ requester,
{
"type": EventTypes.Member,
"content": content,
diff --git a/synapse/http/client.py b/synapse/http/client.py
index ca2f770f5d..9cf797043a 100644
--- a/synapse/http/client.py
+++ b/synapse/http/client.py
@@ -16,7 +16,7 @@ from OpenSSL import SSL
from OpenSSL.SSL import VERIFY_NONE
from synapse.api.errors import (
- CodeMessageException, SynapseError, Codes,
+ CodeMessageException, MatrixCodeMessageException, SynapseError, Codes,
)
from synapse.util.logcontext import preserve_context_over_fn
import synapse.metrics
@@ -145,6 +145,11 @@ class SimpleHttpClient(object):
body = yield preserve_context_over_fn(readBody, response)
+ if 200 <= response.code < 300:
+ defer.returnValue(json.loads(body))
+ else:
+ raise self._exceptionFromFailedRequest(response, body)
+
defer.returnValue(json.loads(body))
@defer.inlineCallbacks
@@ -164,8 +169,11 @@ class SimpleHttpClient(object):
On a non-2xx HTTP response. The response body will be used as the
error message.
"""
- body = yield self.get_raw(uri, args)
- defer.returnValue(json.loads(body))
+ try:
+ body = yield self.get_raw(uri, args)
+ defer.returnValue(json.loads(body))
+ except CodeMessageException as e:
+ raise self._exceptionFromFailedRequest(e.code, e.msg)
@defer.inlineCallbacks
def put_json(self, uri, json_body, args={}):
@@ -246,6 +254,15 @@ class SimpleHttpClient(object):
else:
raise CodeMessageException(response.code, body)
+ def _exceptionFromFailedRequest(self, response, body):
+ try:
+ jsonBody = json.loads(body)
+ errcode = jsonBody['errcode']
+ error = jsonBody['error']
+ return MatrixCodeMessageException(response.code, error, errcode)
+ except (ValueError, KeyError):
+ return CodeMessageException(response.code, body)
+
# XXX: FIXME: This is horribly copy-pasted from matrixfederationclient.
# The two should be factored out.
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index c376ab8fd7..cd388770c8 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -164,6 +164,7 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
else:
msg_handler = self.handlers.message_handler
event, context = yield msg_handler.create_event(
+ requester,
event_dict,
token_id=requester.access_token_id,
txn_id=txn_id,
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index c659004e8d..58b73af7d2 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -60,12 +60,12 @@ class LoggingTransaction(object):
object.__setattr__(self, "database_engine", database_engine)
object.__setattr__(self, "after_callbacks", after_callbacks)
- def call_after(self, callback, *args):
+ def call_after(self, callback, *args, **kwargs):
"""Call the given callback on the main twisted thread after the
transaction has finished. Used to invalidate the caches on the
correct thread.
"""
- self.after_callbacks.append((callback, args))
+ self.after_callbacks.append((callback, args, kwargs))
def __getattr__(self, name):
return getattr(self.txn, name)
@@ -319,8 +319,8 @@ class SQLBaseStore(object):
inner_func, *args, **kwargs
)
finally:
- for after_callback, after_args in after_callbacks:
- after_callback(*after_args)
+ for after_callback, after_args, after_kwargs in after_callbacks:
+ after_callback(*after_args, **after_kwargs)
defer.returnValue(result)
@defer.inlineCallbacks
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index a3790419dd..d946024c9b 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -29,6 +29,7 @@ from synapse.api.constants import EventTypes
from synapse.api.errors import SynapseError
from synapse.state import resolve_events
from synapse.util.caches.descriptors import cached
+from synapse.types import get_domain_from_id
from canonicaljson import encode_canonical_json
from collections import deque, namedtuple, OrderedDict
@@ -49,6 +50,9 @@ logger = logging.getLogger(__name__)
metrics = synapse.metrics.get_metrics_for(__name__)
persist_event_counter = metrics.register_counter("persisted_events")
+event_counter = metrics.register_counter(
+ "persisted_events_sep", labels=["type", "origin_type", "origin_entity"]
+)
def encode_json(json_object):
@@ -371,6 +375,24 @@ class EventsStore(SQLBaseStore):
)
persist_event_counter.inc_by(len(chunk))
+ for room_id, (_, _, new_state) in current_state_for_room.iteritems():
+ self.get_current_state_ids.prefill(
+ (room_id, ), new_state
+ )
+
+ for event, context in chunk:
+ if context.app_service:
+ origin_type = "local"
+ origin_entity = context.app_service.id
+ elif self.hs.is_mine_id(event.sender):
+ origin_type = "local"
+ origin_entity = "*client*"
+ else:
+ origin_type = "remote"
+ origin_entity = get_domain_from_id(event.sender)
+
+ event_counter.inc(event.type, origin_type, origin_entity)
+
@defer.inlineCallbacks
def _calculate_new_extremeties(self, room_id, event_contexts, latest_event_ids):
"""Calculates the new forward extremeties for a room given events to
@@ -419,10 +441,10 @@ class EventsStore(SQLBaseStore):
Assumes that we are only persisting events for one room at a time.
Returns:
- 2-tuple (to_delete, to_insert) where both are state dicts, i.e.
- (type, state_key) -> event_id. `to_delete` are the entries to
+ 3-tuple (to_delete, to_insert, new_state) where both are state dicts,
+ i.e. (type, state_key) -> event_id. `to_delete` are the entries to
first be deleted from current_state_events, `to_insert` are entries
- to insert.
+ to insert. `new_state` is the full set of state.
May return None if there are no changes to be applied.
"""
# Now we need to work out the different state sets for
@@ -529,7 +551,7 @@ class EventsStore(SQLBaseStore):
if ev_id in events_to_insert
}
- defer.returnValue((to_delete, to_insert))
+ defer.returnValue((to_delete, to_insert, current_state))
@defer.inlineCallbacks
def get_event(self, event_id, check_redacted=True,
@@ -682,7 +704,7 @@ class EventsStore(SQLBaseStore):
def _update_current_state_txn(self, txn, state_delta_by_room):
for room_id, current_state_tuple in state_delta_by_room.iteritems():
- to_delete, to_insert = current_state_tuple
+ to_delete, to_insert, _ = current_state_tuple
txn.executemany(
"DELETE FROM current_state_events WHERE event_id = ?",
[(ev_id,) for ev_id in to_delete.itervalues()],
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index a16afa8df5..03981f5d2b 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -227,6 +227,18 @@ class StateStore(SQLBaseStore):
],
)
+ # Prefill the state group cache with this group.
+ # It's fine to use the sequence like this as the state group map
+ # is immutable. (If the map wasn't immutable then this prefill could
+ # race with another update)
+ txn.call_after(
+ self._state_group_cache.update,
+ self._state_group_cache.sequence,
+ key=context.state_group,
+ value=context.current_state_ids,
+ full=True,
+ )
+
self._simple_insert_many_txn(
txn,
table="event_to_state_groups",
diff --git a/tests/storage/event_injector.py b/tests/storage/event_injector.py
index 38556da9a7..024ac15069 100644
--- a/tests/storage/event_injector.py
+++ b/tests/storage/event_injector.py
@@ -27,10 +27,10 @@ class EventInjector:
self.event_builder_factory = hs.get_event_builder_factory()
@defer.inlineCallbacks
- def create_room(self, room):
+ def create_room(self, room, user):
builder = self.event_builder_factory.new({
"type": EventTypes.Create,
- "sender": "",
+ "sender": user.to_string(),
"room_id": room.to_string(),
"content": {},
})
diff --git a/tests/storage/test_events.py b/tests/storage/test_events.py
index 3762b38e37..14443b53bc 100644
--- a/tests/storage/test_events.py
+++ b/tests/storage/test_events.py
@@ -50,7 +50,7 @@ class EventsStoreTestCase(unittest.TestCase):
# Create something to report
room = RoomID.from_string("!abc123:test")
user = UserID.from_string("@raccoonlover:test")
- yield self.event_injector.create_room(room)
+ yield self.event_injector.create_room(room, user)
self.base_event = yield self._get_last_stream_token()
|
