diff --git a/synapse/state.py b/synapse/state.py
index 5dcff27367..9db84c9b5c 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -115,6 +115,8 @@ class StateHandler(object):
is_new = yield self._handle_new_state(new_pdu)
+ logger.debug("is_new: %s %s %s", is_new, new_pdu.pdu_id, new_pdu.origin)
+
if is_new:
yield self.store.update_current_state(
pdu_id=new_pdu.pdu_id,
@@ -134,7 +136,9 @@ class StateHandler(object):
@defer.inlineCallbacks
@log_function
def _handle_new_state(self, new_pdu):
- tree = yield self.store.get_unresolved_state_tree(new_pdu)
+ tree, missing_branch = yield self.store.get_unresolved_state_tree(
+ new_pdu
+ )
new_branch, current_branch = tree
logger.debug(
@@ -142,6 +146,28 @@ class StateHandler(object):
new_branch, current_branch
)
+ 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
+ )
+
+ updated_current = yield self._handle_new_state(new_pdu)
+ defer.returnValue(updated_current)
+
if not current_branch:
# There is no current state
defer.returnValue(True)
@@ -150,84 +176,85 @@ class StateHandler(object):
n = new_branch[-1]
c = current_branch[-1]
- if n.pdu_id == c.pdu_id and n.origin == c.origin:
- # We have all the PDUs we need, so we can just do the conflict
- # resolution.
+ common_ancestor = n.pdu_id == c.pdu_id and n.origin == c.origin
+
+ if common_ancestor:
+ # We found a common ancestor!
if len(current_branch) == 1:
# This is a direct clobber so we can just...
defer.returnValue(True)
- 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 = algo(new_branch, current_branch)
-
- if new_res < curr_res:
- defer.returnValue(False)
- elif new_res > curr_res:
- defer.returnValue(True)
-
- raise Exception("Conflict resolution failed.")
-
else:
- # We need to ask for PDUs.
- missing_prev = max(
- new_branch[-1], current_branch[-1],
- key=lambda x: x.depth
- )
+ # We didn't find a common ancestor. This is probably fine.
+ pass
- if not hasattr(missing_prev, "prev_state_id"):
- # FIXME Hmm
- # temporary fallback
- for algo in conflict_res:
- new_res, curr_res = algo(new_branch, current_branch)
+ result = yield self._do_conflict_res(
+ new_branch, current_branch, common_ancestor
+ )
+ defer.returnValue(result)
- if new_res < curr_res:
- defer.returnValue(False)
- elif new_res > curr_res:
- defer.returnValue(True)
- return
+ @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,
+ ]
- pdu_id = missing_prev.prev_state_id
- origin = missing_prev.prev_state_origin
+ for algo in conflict_res:
+ new_res, curr_res = yield defer.maybeDeferred(
+ algo,
+ new_branch, current_branch, common_ancestor
+ )
- is_missing = yield self.store.get_pdu(pdu_id, origin) is None
+ if new_res < curr_res:
+ defer.returnValue(False)
+ elif new_res > curr_res:
+ defer.returnValue(True)
- if not is_missing:
- raise Exception("Conflict resolution failed.")
+ raise Exception("Conflict resolution failed.")
- yield self._replication.get_pdu(
- destination=missing_prev.origin,
- pdu_origin=origin,
- pdu_id=pdu_id,
- outlier=True
- )
-
- updated_current = yield self._handle_new_state(new_pdu)
- defer.returnValue(updated_current)
+ @defer.inlineCallbacks
+ def _do_power_level_conflict_res(self, new_branch, current_branch,
+ common_ancestor):
+ 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)
+ )
+
+ new_powers = yield defer.gatherResults(
+ new_powers_deferreds,
+ consumeErrors=True
+ )
- def _do_power_level_conflict_res(self, new_branch, current_branch):
- max_power_new = max(
- new_branch[:-1],
- key=lambda t: t.power_level
- ).power_level
+ current_powers = yield defer.gatherResults(
+ current_powers_deferreds,
+ consumeErrors=True
+ )
- max_power_current = max(
- current_branch[:-1],
- key=lambda t: t.power_level
- ).power_level
+ max_power_new = max(new_powers)
+ max_power_current = max(current_powers)
- return (max_power_new, max_power_current)
+ defer.returnValue(
+ (max_power_new, max_power_current)
+ )
- def _do_chain_length_conflict_res(self, new_branch, current_branch):
+ def _do_chain_length_conflict_res(self, new_branch, current_branch,
+ common_ancestor):
return (len(new_branch), len(current_branch))
- def _do_hash_conflict_res(self, new_branch, current_branch):
+ 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])
|
