diff options
Diffstat (limited to 'synapse/state.py')
| -rw-r--r-- | synapse/state.py | 267 |
1 files changed, 110 insertions, 157 deletions
diff --git a/synapse/state.py b/synapse/state.py index bc6b928ec7..f4efc287c9 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -16,11 +16,14 @@ from twisted.internet import defer -from synapse.federation.pdu_codec import encode_event_id, decode_event_id from synapse.util.logutils import log_function +from synapse.util.async import run_on_reactor + +from synapse.types import EventID from collections import namedtuple +import copy import logging import hashlib @@ -35,13 +38,14 @@ KeyStateTuple = namedtuple("KeyStateTuple", ("context", "type", "state_key")) class StateHandler(object): - """ Repsonsible for doing state conflict resolution. + """ Responsible for doing state conflict resolution. """ def __init__(self, hs): self.store = hs.get_datastore() self._replication = hs.get_replication_layer() self.server_name = hs.hostname + self.hs = hs @defer.inlineCallbacks @log_function @@ -50,7 +54,7 @@ class StateHandler(object): to update the state and b) works out what the prev_state should be. Returns: - Deferred: Resolved with a boolean indicating if we succesfully + Deferred: Resolved with a boolean indicating if we successfully updated the state. Raised: @@ -61,200 +65,149 @@ class StateHandler(object): if not hasattr(event, "state_key"): return - key = KeyStateTuple( - event.room_id, - event.type, - _get_state_key_from_event(event) - ) - # Now I need to fill out the prev state and work out if it has auth # (w.r.t. to power levels) snapshot.fill_out_prev_events(event) + yield self.annotate_state_groups(event) - current_state = snapshot.prev_state_pdu - - if current_state: - event.prev_state = encode_event_id( - current_state.pdu_id, current_state.origin + if event.old_state_events: + current_state = event.old_state_events.get( + (event.type, event.state_key) ) - # TODO check current_state to see if the min power level is less - # than the power level of the user - # power_level = self._get_power_level_for_event(event) - - pdu_id, origin = decode_event_id(event.event_id, self.server_name) - - yield self.store.update_current_state( - pdu_id=pdu_id, - origin=origin, - context=key.context, - pdu_type=key.type, - state_key=key.state_key - ) + if current_state: + event.prev_state = current_state.event_id defer.returnValue(True) @defer.inlineCallbacks @log_function - def handle_new_state(self, new_pdu): - """ Apply conflict resolution to `new_pdu`. - - This should be called on every new state pdu, regardless of whether or - not there is a conflict. - - This function is safe against the race of it getting called with two - `PDU`s trying to update the same state. - """ - - # This needs to be done in a transaction. + def annotate_state_groups(self, event, old_state=None): + yield run_on_reactor() - is_new = yield self._handle_new_state(new_pdu) + if old_state: + event.state_group = None + event.old_state_events = { + (s.type, s.state_key): s for s in old_state + } + event.state_events = event.old_state_events - logger.debug("is_new: %s %s %s", is_new, new_pdu.pdu_id, new_pdu.origin) + if hasattr(event, "state_key"): + event.state_events[(event.type, event.state_key)] = event - if is_new: - yield self.store.update_current_state( - pdu_id=new_pdu.pdu_id, - origin=new_pdu.origin, - context=new_pdu.context, - pdu_type=new_pdu.pdu_type, - state_key=new_pdu.state_key - ) - - defer.returnValue(is_new) - - def _get_power_level_for_event(self, event): - # return self._persistence.get_power_level_for_user(event.room_id, - # event.sender) - return event.power_level + defer.returnValue(False) + return - @defer.inlineCallbacks - @log_function - def _handle_new_state(self, new_pdu): - tree, missing_branch = yield self.store.get_unresolved_state_tree( - new_pdu - ) - new_branch, current_branch = tree + if hasattr(event, "outlier") and event.outlier: + event.state_group = None + event.old_state_events = None + event.state_events = {} + defer.returnValue(False) + return - logger.debug( - "_handle_new_state new=%s, current=%s", - new_branch, current_branch + new_state = yield self.resolve_state_groups( + [e for e, _ in event.prev_events] ) - if missing_branch is not None: - # We're missing some PDUs. Fetch them. - # TODO (erikj): Limit this. - missing_prev = tree[missing_branch][-1] - - pdu_id = missing_prev.prev_state_id - origin = missing_prev.prev_state_origin - - is_missing = yield self.store.get_pdu(pdu_id, origin) is None - if not is_missing: - raise Exception("Conflict resolution failed") - - yield self._replication.get_pdu( - destination=missing_prev.origin, - pdu_origin=origin, - pdu_id=pdu_id, - outlier=True - ) + event.old_state_events = copy.deepcopy(new_state) - updated_current = yield self._handle_new_state(new_pdu) - defer.returnValue(updated_current) + if hasattr(event, "state_key"): + new_state[(event.type, event.state_key)] = event - if not current_branch: - # There is no current state - defer.returnValue(True) - return + event.state_group = None + event.state_events = new_state - n = new_branch[-1] - c = current_branch[-1] + defer.returnValue(hasattr(event, "state_key")) - common_ancestor = n.pdu_id == c.pdu_id and n.origin == c.origin + @defer.inlineCallbacks + def get_current_state(self, room_id, event_type=None, state_key=""): + events = yield self.store.get_latest_events_in_room(room_id) - if common_ancestor: - # We found a common ancestor! + event_ids = [ + e_id + for e_id, _, _ in events + ] - if len(current_branch) == 1: - # This is a direct clobber so we can just... - defer.returnValue(True) + res = yield self.resolve_state_groups(event_ids) - else: - # We didn't find a common ancestor. This is probably fine. - pass + if event_type: + defer.returnValue(res.get((event_type, state_key))) + return - result = yield self._do_conflict_res( - new_branch, current_branch, common_ancestor - ) - defer.returnValue(result) + defer.returnValue(res.values()) @defer.inlineCallbacks - def _do_conflict_res(self, new_branch, current_branch, common_ancestor): - conflict_res = [ - self._do_power_level_conflict_res, - self._do_chain_length_conflict_res, - self._do_hash_conflict_res, - ] - - for algo in conflict_res: - new_res, curr_res = yield defer.maybeDeferred( - algo, - new_branch, current_branch, common_ancestor - ) - - if new_res < curr_res: - defer.returnValue(False) - elif new_res > curr_res: - defer.returnValue(True) + @log_function + def resolve_state_groups(self, event_ids): + state_groups = yield self.store.get_state_groups( + event_ids + ) - raise Exception("Conflict resolution failed.") + state = {} + for group in state_groups: + for s in group.state: + state.setdefault( + (s.type, s.state_key), + {} + )[s.event_id] = s + + unconflicted_state = { + k: v.values()[0] for k, v in state.items() + if len(v.values()) == 1 + } + + conflicted_state = { + k: v.values() + for k, v in state.items() + if len(v.values()) > 1 + } + + try: + new_state = {} + new_state.update(unconflicted_state) + for key, events in conflicted_state.items(): + new_state[key] = yield self._resolve_state_events(events) + except: + logger.exception("Failed to resolve state") + raise + + defer.returnValue(new_state) @defer.inlineCallbacks - def _do_power_level_conflict_res(self, new_branch, current_branch, - common_ancestor): + @log_function + def _resolve_state_events(self, events): + curr_events = events + new_powers_deferreds = [] - for e in new_branch[:-1] if common_ancestor else new_branch: - if hasattr(e, "user_id"): - new_powers_deferreds.append( - self.store.get_power_level(e.context, e.user_id) - ) - - current_powers_deferreds = [] - for e in current_branch[:-1] if common_ancestor else current_branch: - if hasattr(e, "user_id"): - current_powers_deferreds.append( - self.store.get_power_level(e.context, e.user_id) - ) + for e in curr_events: + new_powers_deferreds.append( + self.store.get_power_level(e.room_id, e.user_id) + ) new_powers = yield defer.gatherResults( new_powers_deferreds, consumeErrors=True ) - current_powers = yield defer.gatherResults( - current_powers_deferreds, - consumeErrors=True - ) - - max_power_new = max(new_powers) - max_power_current = max(current_powers) + max_power = max([int(p) for p in new_powers]) - defer.returnValue( - (max_power_new, max_power_current) - ) - - def _do_chain_length_conflict_res(self, new_branch, current_branch, - common_ancestor): - return (len(new_branch), len(current_branch)) + curr_events = [ + z[0] for z in zip(curr_events, new_powers) + if int(z[1]) == max_power + ] - def _do_hash_conflict_res(self, new_branch, current_branch, - common_ancestor): - new_str = "".join([p.pdu_id + p.origin for p in new_branch]) - c_str = "".join([p.pdu_id + p.origin for p in current_branch]) + if not curr_events: + raise RuntimeError("Max didn't get a max?") + elif len(curr_events) == 1: + defer.returnValue(curr_events[0]) - return ( - hashlib.sha1(new_str).hexdigest(), - hashlib.sha1(c_str).hexdigest() + # TODO: For now, just choose the one with the largest event_id. + defer.returnValue( + sorted( + curr_events, + key=lambda e: hashlib.sha1( + e.event_id + e.user_id + e.room_id + e.type + ).hexdigest() + )[0] ) |