We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
run_as_background_process doesn't catch & log exceptions 
(plus if it does raise in a loopingcall, we throw away the exception)

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `synapse/metrics/background_process_metrics.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # Copyright 2018 New Vector Ltd
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 import threading
17 
18 import six
19 
20 from prometheus_client.core import REGISTRY, Counter, GaugeMetricFamily
21 
22 from twisted.internet import defer
23 
24 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
25 
26 _background_process_start_count = Counter(
27     "synapse_background_process_start_count",
28     "Number of background processes started",
29     ["name"],
30 )
31 
32 # we set registry=None in all of these to stop them getting registered with
33 # the default registry. Instead we collect them all via the CustomCollector,
34 # which ensures that we can update them before they are collected.
35 #
36 _background_process_ru_utime = Counter(
37     "synapse_background_process_ru_utime_seconds",
38     "User CPU time used by background processes, in seconds",
39     ["name"],
40     registry=None,
41 )
42 
43 _background_process_ru_stime = Counter(
44     "synapse_background_process_ru_stime_seconds",
45     "System CPU time used by background processes, in seconds",
46     ["name"],
47     registry=None,
48 )
49 
50 _background_process_db_txn_count = Counter(
51     "synapse_background_process_db_txn_count",
52     "Number of database transactions done by background processes",
53     ["name"],
54     registry=None,
55 )
56 
57 _background_process_db_txn_duration = Counter(
58     "synapse_background_process_db_txn_duration_seconds",
59     ("Seconds spent by background processes waiting for database "
60      "transactions, excluding scheduling time"),
61     ["name"],
62     registry=None,
63 )
64 
65 _background_process_db_sched_duration = Counter(
66     "synapse_background_process_db_sched_duration_seconds",
67     "Seconds spent by background processes waiting for database connections",
68     ["name"],
69     registry=None,
70 )
71 
72 # map from description to a counter, so that we can name our logcontexts
73 # incrementally. (It actually duplicates _background_process_start_count, but
74 # it's much simpler to do so than to try to combine them.)
75 _background_process_counts = dict()  # type: dict[str, int]
76 
77 # map from description to the currently running background processes.
78 #
79 # it's kept as a dict of sets rather than a big set so that we can keep track
80 # of process descriptions that no longer have any active processes.
81 _background_processes = dict()  # type: dict[str, set[_BackgroundProcess]]
82 
83 # A lock that covers the above dicts
84 _bg_metrics_lock = threading.Lock()
85 
86 
87 class _Collector(object):
88     """A custom metrics collector for the background process metrics.
89 
90     Ensures that all of the metrics are up-to-date with any in-flight processes
91     before they are returned.
92     """
93     def collect(self):
94         background_process_in_flight_count = GaugeMetricFamily(
95             "synapse_background_process_in_flight_count",
96             "Number of background processes in flight",
97             labels=["name"],
98         )
99 
100         # We copy the dict so that it doesn't change from underneath us
101         with _bg_metrics_lock:
102             _background_processes_copy = dict(_background_processes)
103 
104         for desc, processes in six.iteritems(_background_processes_copy):
105             background_process_in_flight_count.add_metric(
106                 (desc,), len(processes),
107             )
108             for process in processes:
109                 process.update_metrics()
110 
111         yield background_process_in_flight_count
112 
113         # now we need to run collect() over each of the static Counters, and
114         # yield each metric they return.
115         for m in (
116                 _background_process_ru_utime,
117                 _background_process_ru_stime,
118                 _background_process_db_txn_count,
119                 _background_process_db_txn_duration,
120                 _background_process_db_sched_duration,
121         ):
122             for r in m.collect():
123                 yield r
124 
125 
126 REGISTRY.register(_Collector())
127 
128 
129 class _BackgroundProcess(object):
130     def __init__(self, desc, ctx):
131         self.desc = desc
132         self._context = ctx
133         self._reported_stats = None
134 
135     def update_metrics(self):
136         """Updates the metrics with values from this process."""
137         new_stats = self._context.get_resource_usage()
138         if self._reported_stats is None:
139             diff = new_stats
140         else:
141             diff = new_stats - self._reported_stats
142         self._reported_stats = new_stats
143 
144         _background_process_ru_utime.labels(self.desc).inc(diff.ru_utime)
145         _background_process_ru_stime.labels(self.desc).inc(diff.ru_stime)
146         _background_process_db_txn_count.labels(self.desc).inc(
147             diff.db_txn_count,
148         )
149         _background_process_db_txn_duration.labels(self.desc).inc(
150             diff.db_txn_duration_sec,
151         )
152         _background_process_db_sched_duration.labels(self.desc).inc(
153             diff.db_sched_duration_sec,
154         )
155 
156 
157 def run_as_background_process(desc, func, *args, **kwargs):
158     """Run the given function in its own logcontext, with resource metrics
159 
160     This should be used to wrap processes which are fired off to run in the
161     background, instead of being associated with a particular request.
162 
163     It returns a Deferred which completes when the function completes, but it doesn't
164     follow the synapse logcontext rules, which makes it appropriate for passing to
165     clock.looping_call and friends (or for firing-and-forgetting in the middle of a
166     normal synapse inlineCallbacks function).
167 
168     Args:
169         desc (str): a description for this background process type
170         func: a function, which may return a Deferred
171         args: positional args for func
172         kwargs: keyword args for func
173 
174     Returns: Deferred which returns the result of func, but note that it does not
175         follow the synapse logcontext rules.
176     """
177     @defer.inlineCallbacks
178     def run():
179         with _bg_metrics_lock:
180             count = _background_process_counts.get(desc, 0)
181             _background_process_counts[desc] = count + 1
182 
183         _background_process_start_count.labels(desc).inc()
184 
185         with LoggingContext(desc) as context:
186             context.request = "%s-%i" % (desc, count)
187             proc = _BackgroundProcess(desc, context)
188 
189             with _bg_metrics_lock:
190                 _background_processes.setdefault(desc, set()).add(proc)
191 
192             try:
193                 yield func(*args, **kwargs)
194             finally:
195                 proc.update_metrics()
196 
197                 with _bg_metrics_lock:
198                     _background_processes[desc].remove(proc)
199 
200     with PreserveLoggingContext():
201         return run()
202
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/synapse/metrics/background_process_metrics.py b/synapse/metrics/background_process_metrics.py
--- a/synapse/metrics/background_process_metrics.py
+++ b/synapse/metrics/background_process_metrics.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 import threading
 
 import six
@@ -23,6 +24,9 @@
 
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
 
+logger = logging.getLogger(__name__)
+
+
 _background_process_start_count = Counter(
     "synapse_background_process_start_count",
     "Number of background processes started",
@@ -191,6 +195,8 @@
 
             try:
                 yield func(*args, **kwargs)
+            except Exception:
+                logger.exception("Background process '%s' threw an exception", desc)
             finally:
                 proc.update_metrics()

{"golden_diff": "diff --git a/synapse/metrics/background_process_metrics.py b/synapse/metrics/background_process_metrics.py\n--- a/synapse/metrics/background_process_metrics.py\n+++ b/synapse/metrics/background_process_metrics.py\n@@ -13,6 +13,7 @@\n # See the License for the specific language governing permissions and\n # limitations under the License.\n \n+import logging\n import threading\n \n import six\n@@ -23,6 +24,9 @@\n \n from synapse.util.logcontext import LoggingContext, PreserveLoggingContext\n \n+logger = logging.getLogger(__name__)\n+\n+\n _background_process_start_count = Counter(\n     \"synapse_background_process_start_count\",\n     \"Number of background processes started\",\n@@ -191,6 +195,8 @@\n \n             try:\n                 yield func(*args, **kwargs)\n+            except Exception:\n+                logger.exception(\"Background process '%s' threw an exception\", desc)\n             finally:\n                 proc.update_metrics()\n", "issue": "run_as_background_process doesn't catch & log exceptions \n(plus if it does raise in a loopingcall, we throw away the exception)\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2018 New Vector Ltd\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport threading\n\nimport six\n\nfrom prometheus_client.core import REGISTRY, Counter, GaugeMetricFamily\n\nfrom twisted.internet import defer\n\nfrom synapse.util.logcontext import LoggingContext, PreserveLoggingContext\n\n_background_process_start_count = Counter(\n    \"synapse_background_process_start_count\",\n    \"Number of background processes started\",\n    [\"name\"],\n)\n\n# we set registry=None in all of these to stop them getting registered with\n# the default registry. Instead we collect them all via the CustomCollector,\n# which ensures that we can update them before they are collected.\n#\n_background_process_ru_utime = Counter(\n    \"synapse_background_process_ru_utime_seconds\",\n    \"User CPU time used by background processes, in seconds\",\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_ru_stime = Counter(\n    \"synapse_background_process_ru_stime_seconds\",\n    \"System CPU time used by background processes, in seconds\",\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_db_txn_count = Counter(\n    \"synapse_background_process_db_txn_count\",\n    \"Number of database transactions done by background processes\",\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_db_txn_duration = Counter(\n    \"synapse_background_process_db_txn_duration_seconds\",\n    (\"Seconds spent by background processes waiting for database \"\n     \"transactions, excluding scheduling time\"),\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_db_sched_duration = Counter(\n    \"synapse_background_process_db_sched_duration_seconds\",\n    \"Seconds spent by background processes waiting for database connections\",\n    [\"name\"],\n    registry=None,\n)\n\n# map from description to a counter, so that we can name our logcontexts\n# incrementally. (It actually duplicates _background_process_start_count, but\n# it's much simpler to do so than to try to combine them.)\n_background_process_counts = dict()  # type: dict[str, int]\n\n# map from description to the currently running background processes.\n#\n# it's kept as a dict of sets rather than a big set so that we can keep track\n# of process descriptions that no longer have any active processes.\n_background_processes = dict()  # type: dict[str, set[_BackgroundProcess]]\n\n# A lock that covers the above dicts\n_bg_metrics_lock = threading.Lock()\n\n\nclass _Collector(object):\n    \"\"\"A custom metrics collector for the background process metrics.\n\n    Ensures that all of the metrics are up-to-date with any in-flight processes\n    before they are returned.\n    \"\"\"\n    def collect(self):\n        background_process_in_flight_count = GaugeMetricFamily(\n            \"synapse_background_process_in_flight_count\",\n            \"Number of background processes in flight\",\n            labels=[\"name\"],\n        )\n\n        # We copy the dict so that it doesn't change from underneath us\n        with _bg_metrics_lock:\n            _background_processes_copy = dict(_background_processes)\n\n        for desc, processes in six.iteritems(_background_processes_copy):\n            background_process_in_flight_count.add_metric(\n                (desc,), len(processes),\n            )\n            for process in processes:\n                process.update_metrics()\n\n        yield background_process_in_flight_count\n\n        # now we need to run collect() over each of the static Counters, and\n        # yield each metric they return.\n        for m in (\n                _background_process_ru_utime,\n                _background_process_ru_stime,\n                _background_process_db_txn_count,\n                _background_process_db_txn_duration,\n                _background_process_db_sched_duration,\n        ):\n            for r in m.collect():\n                yield r\n\n\nREGISTRY.register(_Collector())\n\n\nclass _BackgroundProcess(object):\n    def __init__(self, desc, ctx):\n        self.desc = desc\n        self._context = ctx\n        self._reported_stats = None\n\n    def update_metrics(self):\n        \"\"\"Updates the metrics with values from this process.\"\"\"\n        new_stats = self._context.get_resource_usage()\n        if self._reported_stats is None:\n            diff = new_stats\n        else:\n            diff = new_stats - self._reported_stats\n        self._reported_stats = new_stats\n\n        _background_process_ru_utime.labels(self.desc).inc(diff.ru_utime)\n        _background_process_ru_stime.labels(self.desc).inc(diff.ru_stime)\n        _background_process_db_txn_count.labels(self.desc).inc(\n            diff.db_txn_count,\n        )\n        _background_process_db_txn_duration.labels(self.desc).inc(\n            diff.db_txn_duration_sec,\n        )\n        _background_process_db_sched_duration.labels(self.desc).inc(\n            diff.db_sched_duration_sec,\n        )\n\n\ndef run_as_background_process(desc, func, *args, **kwargs):\n    \"\"\"Run the given function in its own logcontext, with resource metrics\n\n    This should be used to wrap processes which are fired off to run in the\n    background, instead of being associated with a particular request.\n\n    It returns a Deferred which completes when the function completes, but it doesn't\n    follow the synapse logcontext rules, which makes it appropriate for passing to\n    clock.looping_call and friends (or for firing-and-forgetting in the middle of a\n    normal synapse inlineCallbacks function).\n\n    Args:\n        desc (str): a description for this background process type\n        func: a function, which may return a Deferred\n        args: positional args for func\n        kwargs: keyword args for func\n\n    Returns: Deferred which returns the result of func, but note that it does not\n        follow the synapse logcontext rules.\n    \"\"\"\n    @defer.inlineCallbacks\n    def run():\n        with _bg_metrics_lock:\n            count = _background_process_counts.get(desc, 0)\n            _background_process_counts[desc] = count + 1\n\n        _background_process_start_count.labels(desc).inc()\n\n        with LoggingContext(desc) as context:\n            context.request = \"%s-%i\" % (desc, count)\n            proc = _BackgroundProcess(desc, context)\n\n            with _bg_metrics_lock:\n                _background_processes.setdefault(desc, set()).add(proc)\n\n            try:\n                yield func(*args, **kwargs)\n            finally:\n                proc.update_metrics()\n\n                with _bg_metrics_lock:\n                    _background_processes[desc].remove(proc)\n\n    with PreserveLoggingContext():\n        return run()\n", "path": "synapse/metrics/background_process_metrics.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2018 New Vector Ltd\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nimport threading\n\nimport six\n\nfrom prometheus_client.core import REGISTRY, Counter, GaugeMetricFamily\n\nfrom twisted.internet import defer\n\nfrom synapse.util.logcontext import LoggingContext, PreserveLoggingContext\n\nlogger = logging.getLogger(__name__)\n\n\n_background_process_start_count = Counter(\n    \"synapse_background_process_start_count\",\n    \"Number of background processes started\",\n    [\"name\"],\n)\n\n# we set registry=None in all of these to stop them getting registered with\n# the default registry. Instead we collect them all via the CustomCollector,\n# which ensures that we can update them before they are collected.\n#\n_background_process_ru_utime = Counter(\n    \"synapse_background_process_ru_utime_seconds\",\n    \"User CPU time used by background processes, in seconds\",\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_ru_stime = Counter(\n    \"synapse_background_process_ru_stime_seconds\",\n    \"System CPU time used by background processes, in seconds\",\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_db_txn_count = Counter(\n    \"synapse_background_process_db_txn_count\",\n    \"Number of database transactions done by background processes\",\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_db_txn_duration = Counter(\n    \"synapse_background_process_db_txn_duration_seconds\",\n    (\"Seconds spent by background processes waiting for database \"\n     \"transactions, excluding scheduling time\"),\n    [\"name\"],\n    registry=None,\n)\n\n_background_process_db_sched_duration = Counter(\n    \"synapse_background_process_db_sched_duration_seconds\",\n    \"Seconds spent by background processes waiting for database connections\",\n    [\"name\"],\n    registry=None,\n)\n\n# map from description to a counter, so that we can name our logcontexts\n# incrementally. (It actually duplicates _background_process_start_count, but\n# it's much simpler to do so than to try to combine them.)\n_background_process_counts = dict()  # type: dict[str, int]\n\n# map from description to the currently running background processes.\n#\n# it's kept as a dict of sets rather than a big set so that we can keep track\n# of process descriptions that no longer have any active processes.\n_background_processes = dict()  # type: dict[str, set[_BackgroundProcess]]\n\n# A lock that covers the above dicts\n_bg_metrics_lock = threading.Lock()\n\n\nclass _Collector(object):\n    \"\"\"A custom metrics collector for the background process metrics.\n\n    Ensures that all of the metrics are up-to-date with any in-flight processes\n    before they are returned.\n    \"\"\"\n    def collect(self):\n        background_process_in_flight_count = GaugeMetricFamily(\n            \"synapse_background_process_in_flight_count\",\n            \"Number of background processes in flight\",\n            labels=[\"name\"],\n        )\n\n        # We copy the dict so that it doesn't change from underneath us\n        with _bg_metrics_lock:\n            _background_processes_copy = dict(_background_processes)\n\n        for desc, processes in six.iteritems(_background_processes_copy):\n            background_process_in_flight_count.add_metric(\n                (desc,), len(processes),\n            )\n            for process in processes:\n                process.update_metrics()\n\n        yield background_process_in_flight_count\n\n        # now we need to run collect() over each of the static Counters, and\n        # yield each metric they return.\n        for m in (\n                _background_process_ru_utime,\n                _background_process_ru_stime,\n                _background_process_db_txn_count,\n                _background_process_db_txn_duration,\n                _background_process_db_sched_duration,\n        ):\n            for r in m.collect():\n                yield r\n\n\nREGISTRY.register(_Collector())\n\n\nclass _BackgroundProcess(object):\n    def __init__(self, desc, ctx):\n        self.desc = desc\n        self._context = ctx\n        self._reported_stats = None\n\n    def update_metrics(self):\n        \"\"\"Updates the metrics with values from this process.\"\"\"\n        new_stats = self._context.get_resource_usage()\n        if self._reported_stats is None:\n            diff = new_stats\n        else:\n            diff = new_stats - self._reported_stats\n        self._reported_stats = new_stats\n\n        _background_process_ru_utime.labels(self.desc).inc(diff.ru_utime)\n        _background_process_ru_stime.labels(self.desc).inc(diff.ru_stime)\n        _background_process_db_txn_count.labels(self.desc).inc(\n            diff.db_txn_count,\n        )\n        _background_process_db_txn_duration.labels(self.desc).inc(\n            diff.db_txn_duration_sec,\n        )\n        _background_process_db_sched_duration.labels(self.desc).inc(\n            diff.db_sched_duration_sec,\n        )\n\n\ndef run_as_background_process(desc, func, *args, **kwargs):\n    \"\"\"Run the given function in its own logcontext, with resource metrics\n\n    This should be used to wrap processes which are fired off to run in the\n    background, instead of being associated with a particular request.\n\n    It returns a Deferred which completes when the function completes, but it doesn't\n    follow the synapse logcontext rules, which makes it appropriate for passing to\n    clock.looping_call and friends (or for firing-and-forgetting in the middle of a\n    normal synapse inlineCallbacks function).\n\n    Args:\n        desc (str): a description for this background process type\n        func: a function, which may return a Deferred\n        args: positional args for func\n        kwargs: keyword args for func\n\n    Returns: Deferred which returns the result of func, but note that it does not\n        follow the synapse logcontext rules.\n    \"\"\"\n    @defer.inlineCallbacks\n    def run():\n        with _bg_metrics_lock:\n            count = _background_process_counts.get(desc, 0)\n            _background_process_counts[desc] = count + 1\n\n        _background_process_start_count.labels(desc).inc()\n\n        with LoggingContext(desc) as context:\n            context.request = \"%s-%i\" % (desc, count)\n            proc = _BackgroundProcess(desc, context)\n\n            with _bg_metrics_lock:\n                _background_processes.setdefault(desc, set()).add(proc)\n\n            try:\n                yield func(*args, **kwargs)\n            except Exception:\n                logger.exception(\"Background process '%s' threw an exception\", desc)\n            finally:\n                proc.update_metrics()\n\n                with _bg_metrics_lock:\n                    _background_processes[desc].remove(proc)\n\n    with PreserveLoggingContext():\n        return run()\n", "path": "synapse/metrics/background_process_metrics.py"}]}