Merge branch 'develop' into matrix-org-hotfixes

6 files changed, 68 insertions, 15 deletions

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index f00d59e701..a8034bddc6 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -211,8 +211,9 @@ class FederationServer(FederationBase):
                     edu.content
                 )
 
-            for failure in getattr(transaction, "pdu_failures", []):
-                logger.info("Got failure %r", failure)
+        pdu_failures = getattr(transaction, "pdu_failures", [])
+        for failure in pdu_failures:
+            logger.info("Got failure %r", failure)
 
         response = {
             "pdus": pdu_results,
diff --git a/synapse/handlers/groups_local.py b/synapse/handlers/groups_local.py
index 50e40548c2..3b676d46bd 100644
--- a/synapse/handlers/groups_local.py
+++ b/synapse/handlers/groups_local.py
@@ -109,7 +109,7 @@ class GroupsLocalHandler(object):
             valid_users = []
             for entry in chunk:
                 g_user_id = entry["user_id"]
-                attestation = entry.pop("attestation")
+                attestation = entry.pop("attestation", {})
                 try:
                     if get_domain_from_id(g_user_id) != group_server_name:
                         yield self.attestations.verify_attestation(
@@ -202,7 +202,7 @@ class GroupsLocalHandler(object):
         valid_entries = []
         for entry in chunk:
             g_user_id = entry["user_id"]
-            attestation = entry.pop("attestation")
+            attestation = entry.pop("attestation", {})
             try:
                 if get_domain_from_id(g_user_id) != group_server_name:
                     yield self.attestations.verify_attestation(
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index fbf88b46ef..28792788d9 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014 - 2016 OpenMarket Ltd
+# Copyright 2017 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -325,9 +326,12 @@ class MessageHandler(BaseHandler):
             txn_id=txn_id
         )
 
-        if self.spam_checker.check_event_for_spam(event):
+        spam_error = self.spam_checker.check_event_for_spam(event)
+        if spam_error:
+            if not isinstance(spam_error, basestring):
+                spam_error = "Spam is not permitted here"
             raise SynapseError(
-                403, "Spam is not permitted here", Codes.FORBIDDEN
+                403, spam_error, Codes.FORBIDDEN
             )
 
         yield self.send_nonmember_event(
diff --git a/synapse/util/async.py b/synapse/util/async.py
index bb252f75d7..0fd5b42523 100644
--- a/synapse/util/async.py
+++ b/synapse/util/async.py
@@ -203,7 +203,26 @@ class Linearizer(object):
             except:
                 logger.exception("Unexpected exception in Linearizer")
 
-        logger.info("Acquired linearizer lock %r for key %r", self.name, key)
+            logger.info("Acquired linearizer lock %r for key %r", self.name,
+                        key)
+
+            # if the code holding the lock completes synchronously, then it
+            # will recursively run the next claimant on the list. That can
+            # relatively rapidly lead to stack exhaustion. This is essentially
+            # the same problem as http://twistedmatrix.com/trac/ticket/9304.
+            #
+            # In order to break the cycle, we add a cheeky sleep(0) here to
+            # ensure that we fall back to the reactor between each iteration.
+            #
+            # (There's no particular need for it to happen before we return
+            # the context manager, but it needs to happen while we hold the
+            # lock, and the context manager's exit code must be synchronous,
+            # so actually this is the only sensible place.
+            yield run_on_reactor()
+
+        else:
+            logger.info("Acquired uncontended linearizer lock %r for key %r",
+                        self.name, key)
 
         @contextmanager
         def _ctx_manager():
@@ -211,7 +230,8 @@ class Linearizer(object):
                 yield
             finally:
                 logger.info("Releasing linearizer lock %r for key %r", self.name, key)
-                new_defer.callback(None)
+                with PreserveLoggingContext():
+                    new_defer.callback(None)
                 current_d = self.key_to_defer.get(key)
                 if current_d is new_defer:
                     self.key_to_defer.pop(key, None)
diff --git a/synapse/util/logformatter.py b/synapse/util/logformatter.py
index 60504162e9..cdbc4bffd7 100644
--- a/synapse/util/logformatter.py
+++ b/synapse/util/logformatter.py
@@ -33,9 +33,17 @@ class LogFormatter(logging.Formatter):
 
     def formatException(self, ei):
         sio = StringIO.StringIO()
-        sio.write("Capture point (most recent call last):\n")
-        traceback.print_stack(ei[2].tb_frame.f_back, None, sio)
-        traceback.print_exception(ei[0], ei[1], ei[2], None, sio)
+        (typ, val, tb) = ei
+
+        # log the stack above the exception capture point if possible, but
+        # check that we actually have an f_back attribute to work around
+        # https://twistedmatrix.com/trac/ticket/9305
+
+        if tb and hasattr(tb.tb_frame, 'f_back'):
+            sio.write("Capture point (most recent call last):\n")
+            traceback.print_stack(tb.tb_frame.f_back, None, sio)
+
+        traceback.print_exception(typ, val, tb, None, sio)
         s = sio.getvalue()
         sio.close()
         if s[-1:] == "\n":
diff --git a/tests/util/test_linearizer.py b/tests/util/test_linearizer.py
index afcba482f9..793a88e462 100644
--- a/tests/util/test_linearizer.py
+++ b/tests/util/test_linearizer.py
@@ -12,8 +12,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
-
+from synapse.util import async, logcontext
 from tests import unittest
 
 from twisted.internet import defer
@@ -38,7 +37,28 @@ class LinearizerTestCase(unittest.TestCase):
         with cm1:
             self.assertFalse(d2.called)
 
-        self.assertTrue(d2.called)
-
         with (yield d2):
             pass
+
+    def test_lots_of_queued_things(self):
+        # we have one slow thing, and lots of fast things queued up behind it.
+        # it should *not* explode the stack.
+        linearizer = Linearizer()
+
+        @defer.inlineCallbacks
+        def func(i, sleep=False):
+            with logcontext.LoggingContext("func(%s)" % i) as lc:
+                with (yield linearizer.queue("")):
+                    self.assertEqual(
+                        logcontext.LoggingContext.current_context(), lc)
+                    if sleep:
+                        yield async.sleep(0)
+
+                self.assertEqual(
+                    logcontext.LoggingContext.current_context(), lc)
+
+        func(0, sleep=True)
+        for i in xrange(1, 100):
+            func(i)
+
+        return func(1000)