Sort out error handling

1 files changed, 22 insertions, 25 deletions

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 260bdf0ec4..f2c181dde7 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -502,8 +502,8 @@ class EventsStore(SQLBaseStore):
 
     def _do_fetch(self, conn):
         event_list = []
-        try:
-            while True:
+        while True:
+            try:
                 logger.debug("do_fetch getting lock")
                 with self._event_fetch_lock:
                     logger.debug("do_fetch go lock: %r", self._event_fetch_list)
@@ -543,16 +543,16 @@ class EventsStore(SQLBaseStore):
                             except:
                                 logger.exception("Failed to callback")
                 reactor.callFromThread(fire, event_list)
-        except Exception as e:
-            logger.exception("do_fetch")
+            except Exception as e:
+                logger.exception("do_fetch")
 
-            def fire(evs):
-                for _, d in evs:
-                    if not d.called:
-                        d.errback(e)
+                def fire(evs):
+                    for _, d in evs:
+                        if not d.called:
+                            d.errback(e)
 
-            if event_list:
-                reactor.callFromThread(fire, event_list)
+                if event_list:
+                    reactor.callFromThread(fire, event_list)
 
     @defer.inlineCallbacks
     def _enqueue_events(self, events, check_redacted=True,
@@ -561,29 +561,26 @@ class EventsStore(SQLBaseStore):
             defer.returnValue({})
 
         events_d = defer.Deferred()
-        try:
-            logger.debug("enqueueueueue getting lock")
-            with self._event_fetch_lock:
-                logger.debug("enqueue go lock")
-                self._event_fetch_list.append(
-                    (events, events_d)
-                )
+        logger.debug("enqueueueueue getting lock")
+        with self._event_fetch_lock:
+            logger.debug("enqueue go lock")
+            self._event_fetch_list.append(
+                (events, events_d)
+            )
 
+            if self._event_fetch_ongoing < 1:
                 self._event_fetch_ongoing += 1
+                should_start = True
+            else:
+                should_start = False
 
+        if should_start:
             self.runWithConnection(
                 self._do_fetch
             )
 
-        except Exception as e:
-            if not events_d.called:
-                events_d.errback(e)
-
         logger.debug("events_d before")
-        try:
-            rows = yield events_d
-        except:
-            logger.exception("events_d")
+        rows = yield events_d
         logger.debug("events_d after")
 
         res = yield defer.gatherResults(