We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
celery integration RecursionError
Hi there,
I upgraded sentry_sdk to 0.7.0 and started getting RecursionError if there's an issue with celery task. Sentry record doesn't contain any stack trace for that but found that error in my apm system (can attach screenshot only, text data is a real mess there). I'm running celery 4.2.1 on Ubuntu 18. 

<img width="618" alt="2019-02-14 15 04 54" src="https://user-images.githubusercontent.com/7299611/52785969-89ea2180-306a-11e9-9828-56a52f3fddd6.png">


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sentry_sdk/integrations/celery.py`
Content:
```
1 from __future__ import absolute_import
2 
3 import sys
4 
5 from celery.exceptions import (  # type: ignore
6     SoftTimeLimitExceeded,
7     Retry,
8     Ignore,
9     Reject,
10 )
11 
12 from sentry_sdk.hub import Hub
13 from sentry_sdk.utils import capture_internal_exceptions, event_from_exception
14 from sentry_sdk.tracing import SpanContext
15 from sentry_sdk._compat import reraise
16 from sentry_sdk.integrations import Integration
17 from sentry_sdk.integrations.logging import ignore_logger
18 
19 
20 CELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)
21 
22 
23 class CeleryIntegration(Integration):
24     identifier = "celery"
25 
26     def __init__(self, propagate_traces=True):
27         self.propagate_traces = propagate_traces
28 
29     @staticmethod
30     def setup_once():
31         import celery.app.trace as trace  # type: ignore
32 
33         old_build_tracer = trace.build_tracer
34 
35         def sentry_build_tracer(name, task, *args, **kwargs):
36             # Need to patch both methods because older celery sometimes
37             # short-circuits to task.run if it thinks it's safe.
38             task.__call__ = _wrap_task_call(task, task.__call__)
39             task.run = _wrap_task_call(task, task.run)
40             task.apply_async = _wrap_apply_async(task, task.apply_async)
41             return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))
42 
43         trace.build_tracer = sentry_build_tracer
44 
45         _patch_worker_exit()
46 
47         # This logger logs every status of every task that ran on the worker.
48         # Meaning that every task's breadcrumbs are full of stuff like "Task
49         # <foo> raised unexpected <bar>".
50         ignore_logger("celery.worker.job")
51 
52 
53 def _wrap_apply_async(task, f):
54     def apply_async(*args, **kwargs):
55         hub = Hub.current
56         integration = hub.get_integration(CeleryIntegration)
57         if integration is not None and integration.propagate_traces:
58             headers = None
59             for key, value in hub.iter_trace_propagation_headers():
60                 if headers is None:
61                     headers = dict(kwargs.get("headers") or {})
62                 headers[key] = value
63             if headers is not None:
64                 kwargs["headers"] = headers
65         return f(*args, **kwargs)
66 
67     return apply_async
68 
69 
70 def _wrap_tracer(task, f):
71     # Need to wrap tracer for pushing the scope before prerun is sent, and
72     # popping it after postrun is sent.
73     #
74     # This is the reason we don't use signals for hooking in the first place.
75     # Also because in Celery 3, signal dispatch returns early if one handler
76     # crashes.
77     def _inner(*args, **kwargs):
78         hub = Hub.current
79         if hub.get_integration(CeleryIntegration) is None:
80             return f(*args, **kwargs)
81 
82         with hub.push_scope() as scope:
83             scope._name = "celery"
84             scope.clear_breadcrumbs()
85             _continue_trace(args[3].get("headers") or {}, scope)
86             scope.add_event_processor(_make_event_processor(task, *args, **kwargs))
87 
88             return f(*args, **kwargs)
89 
90     return _inner
91 
92 
93 def _continue_trace(headers, scope):
94     if headers:
95         span_context = SpanContext.continue_from_headers(headers)
96     else:
97         span_context = SpanContext.start_trace()
98     scope.set_span_context(span_context)
99 
100 
101 def _wrap_task_call(task, f):
102     # Need to wrap task call because the exception is caught before we get to
103     # see it. Also celery's reported stacktrace is untrustworthy.
104     def _inner(*args, **kwargs):
105         try:
106             return f(*args, **kwargs)
107         except Exception:
108             exc_info = sys.exc_info()
109             with capture_internal_exceptions():
110                 _capture_exception(task, exc_info)
111             reraise(*exc_info)
112 
113     return _inner
114 
115 
116 def _make_event_processor(task, uuid, args, kwargs, request=None):
117     def event_processor(event, hint):
118         with capture_internal_exceptions():
119             event["transaction"] = task.name
120 
121         with capture_internal_exceptions():
122             extra = event.setdefault("extra", {})
123             extra["celery-job"] = {
124                 "task_name": task.name,
125                 "args": args,
126                 "kwargs": kwargs,
127             }
128 
129         if "exc_info" in hint:
130             with capture_internal_exceptions():
131                 if issubclass(hint["exc_info"][0], SoftTimeLimitExceeded):
132                     event["fingerprint"] = [
133                         "celery",
134                         "SoftTimeLimitExceeded",
135                         getattr(task, "name", task),
136                     ]
137 
138         return event
139 
140     return event_processor
141 
142 
143 def _capture_exception(task, exc_info):
144     hub = Hub.current
145 
146     if hub.get_integration(CeleryIntegration) is None:
147         return
148     if isinstance(exc_info[1], CELERY_CONTROL_FLOW_EXCEPTIONS):
149         return
150     if hasattr(task, "throws") and isinstance(exc_info[1], task.throws):
151         return
152 
153     event, hint = event_from_exception(
154         exc_info,
155         client_options=hub.client.options,
156         mechanism={"type": "celery", "handled": False},
157     )
158 
159     hub.capture_event(event, hint=hint)
160 
161 
162 def _patch_worker_exit():
163     # Need to flush queue before worker shutdown because a crashing worker will
164     # call os._exit
165     from billiard.pool import Worker  # type: ignore
166 
167     old_workloop = Worker.workloop
168 
169     def sentry_workloop(*args, **kwargs):
170         try:
171             return old_workloop(*args, **kwargs)
172         finally:
173             with capture_internal_exceptions():
174                 hub = Hub.current
175                 if hub.get_integration(CeleryIntegration) is not None:
176                     hub.flush()
177 
178     Worker.workloop = sentry_workloop
179
```
--- 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/sentry_sdk/integrations/celery.py b/sentry_sdk/integrations/celery.py
--- a/sentry_sdk/integrations/celery.py
+++ b/sentry_sdk/integrations/celery.py
@@ -33,11 +33,18 @@
         old_build_tracer = trace.build_tracer
 
         def sentry_build_tracer(name, task, *args, **kwargs):
-            # Need to patch both methods because older celery sometimes
-            # short-circuits to task.run if it thinks it's safe.
-            task.__call__ = _wrap_task_call(task, task.__call__)
-            task.run = _wrap_task_call(task, task.run)
-            task.apply_async = _wrap_apply_async(task, task.apply_async)
+            if not getattr(task, "_sentry_is_patched", False):
+                # Need to patch both methods because older celery sometimes
+                # short-circuits to task.run if it thinks it's safe.
+                task.__call__ = _wrap_task_call(task, task.__call__)
+                task.run = _wrap_task_call(task, task.run)
+                task.apply_async = _wrap_apply_async(task, task.apply_async)
+
+                # `build_tracer` is apparently called for every task
+                # invocation. Can't wrap every celery task for every invocation
+                # or we will get infinitely nested wrapper functions.
+                task._sentry_is_patched = True
+
             return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))
 
         trace.build_tracer = sentry_build_tracer

{"golden_diff": "diff --git a/sentry_sdk/integrations/celery.py b/sentry_sdk/integrations/celery.py\n--- a/sentry_sdk/integrations/celery.py\n+++ b/sentry_sdk/integrations/celery.py\n@@ -33,11 +33,18 @@\n         old_build_tracer = trace.build_tracer\n \n         def sentry_build_tracer(name, task, *args, **kwargs):\n-            # Need to patch both methods because older celery sometimes\n-            # short-circuits to task.run if it thinks it's safe.\n-            task.__call__ = _wrap_task_call(task, task.__call__)\n-            task.run = _wrap_task_call(task, task.run)\n-            task.apply_async = _wrap_apply_async(task, task.apply_async)\n+            if not getattr(task, \"_sentry_is_patched\", False):\n+                # Need to patch both methods because older celery sometimes\n+                # short-circuits to task.run if it thinks it's safe.\n+                task.__call__ = _wrap_task_call(task, task.__call__)\n+                task.run = _wrap_task_call(task, task.run)\n+                task.apply_async = _wrap_apply_async(task, task.apply_async)\n+\n+                # `build_tracer` is apparently called for every task\n+                # invocation. Can't wrap every celery task for every invocation\n+                # or we will get infinitely nested wrapper functions.\n+                task._sentry_is_patched = True\n+\n             return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))\n \n         trace.build_tracer = sentry_build_tracer\n", "issue": "celery integration RecursionError\nHi there,\r\nI upgraded sentry_sdk to 0.7.0 and started getting RecursionError if there's an issue with celery task. Sentry record doesn't contain any stack trace for that but found that error in my apm system (can attach screenshot only, text data is a real mess there). I'm running celery 4.2.1 on Ubuntu 18. \r\n\r\n<img width=\"618\" alt=\"2019-02-14 15 04 54\" src=\"https://user-images.githubusercontent.com/7299611/52785969-89ea2180-306a-11e9-9828-56a52f3fddd6.png\">\r\n\n", "before_files": [{"content": "from __future__ import absolute_import\n\nimport sys\n\nfrom celery.exceptions import (  # type: ignore\n    SoftTimeLimitExceeded,\n    Retry,\n    Ignore,\n    Reject,\n)\n\nfrom sentry_sdk.hub import Hub\nfrom sentry_sdk.utils import capture_internal_exceptions, event_from_exception\nfrom sentry_sdk.tracing import SpanContext\nfrom sentry_sdk._compat import reraise\nfrom sentry_sdk.integrations import Integration\nfrom sentry_sdk.integrations.logging import ignore_logger\n\n\nCELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)\n\n\nclass CeleryIntegration(Integration):\n    identifier = \"celery\"\n\n    def __init__(self, propagate_traces=True):\n        self.propagate_traces = propagate_traces\n\n    @staticmethod\n    def setup_once():\n        import celery.app.trace as trace  # type: ignore\n\n        old_build_tracer = trace.build_tracer\n\n        def sentry_build_tracer(name, task, *args, **kwargs):\n            # Need to patch both methods because older celery sometimes\n            # short-circuits to task.run if it thinks it's safe.\n            task.__call__ = _wrap_task_call(task, task.__call__)\n            task.run = _wrap_task_call(task, task.run)\n            task.apply_async = _wrap_apply_async(task, task.apply_async)\n            return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))\n\n        trace.build_tracer = sentry_build_tracer\n\n        _patch_worker_exit()\n\n        # This logger logs every status of every task that ran on the worker.\n        # Meaning that every task's breadcrumbs are full of stuff like \"Task\n        # <foo> raised unexpected <bar>\".\n        ignore_logger(\"celery.worker.job\")\n\n\ndef _wrap_apply_async(task, f):\n    def apply_async(*args, **kwargs):\n        hub = Hub.current\n        integration = hub.get_integration(CeleryIntegration)\n        if integration is not None and integration.propagate_traces:\n            headers = None\n            for key, value in hub.iter_trace_propagation_headers():\n                if headers is None:\n                    headers = dict(kwargs.get(\"headers\") or {})\n                headers[key] = value\n            if headers is not None:\n                kwargs[\"headers\"] = headers\n        return f(*args, **kwargs)\n\n    return apply_async\n\n\ndef _wrap_tracer(task, f):\n    # Need to wrap tracer for pushing the scope before prerun is sent, and\n    # popping it after postrun is sent.\n    #\n    # This is the reason we don't use signals for hooking in the first place.\n    # Also because in Celery 3, signal dispatch returns early if one handler\n    # crashes.\n    def _inner(*args, **kwargs):\n        hub = Hub.current\n        if hub.get_integration(CeleryIntegration) is None:\n            return f(*args, **kwargs)\n\n        with hub.push_scope() as scope:\n            scope._name = \"celery\"\n            scope.clear_breadcrumbs()\n            _continue_trace(args[3].get(\"headers\") or {}, scope)\n            scope.add_event_processor(_make_event_processor(task, *args, **kwargs))\n\n            return f(*args, **kwargs)\n\n    return _inner\n\n\ndef _continue_trace(headers, scope):\n    if headers:\n        span_context = SpanContext.continue_from_headers(headers)\n    else:\n        span_context = SpanContext.start_trace()\n    scope.set_span_context(span_context)\n\n\ndef _wrap_task_call(task, f):\n    # Need to wrap task call because the exception is caught before we get to\n    # see it. Also celery's reported stacktrace is untrustworthy.\n    def _inner(*args, **kwargs):\n        try:\n            return f(*args, **kwargs)\n        except Exception:\n            exc_info = sys.exc_info()\n            with capture_internal_exceptions():\n                _capture_exception(task, exc_info)\n            reraise(*exc_info)\n\n    return _inner\n\n\ndef _make_event_processor(task, uuid, args, kwargs, request=None):\n    def event_processor(event, hint):\n        with capture_internal_exceptions():\n            event[\"transaction\"] = task.name\n\n        with capture_internal_exceptions():\n            extra = event.setdefault(\"extra\", {})\n            extra[\"celery-job\"] = {\n                \"task_name\": task.name,\n                \"args\": args,\n                \"kwargs\": kwargs,\n            }\n\n        if \"exc_info\" in hint:\n            with capture_internal_exceptions():\n                if issubclass(hint[\"exc_info\"][0], SoftTimeLimitExceeded):\n                    event[\"fingerprint\"] = [\n                        \"celery\",\n                        \"SoftTimeLimitExceeded\",\n                        getattr(task, \"name\", task),\n                    ]\n\n        return event\n\n    return event_processor\n\n\ndef _capture_exception(task, exc_info):\n    hub = Hub.current\n\n    if hub.get_integration(CeleryIntegration) is None:\n        return\n    if isinstance(exc_info[1], CELERY_CONTROL_FLOW_EXCEPTIONS):\n        return\n    if hasattr(task, \"throws\") and isinstance(exc_info[1], task.throws):\n        return\n\n    event, hint = event_from_exception(\n        exc_info,\n        client_options=hub.client.options,\n        mechanism={\"type\": \"celery\", \"handled\": False},\n    )\n\n    hub.capture_event(event, hint=hint)\n\n\ndef _patch_worker_exit():\n    # Need to flush queue before worker shutdown because a crashing worker will\n    # call os._exit\n    from billiard.pool import Worker  # type: ignore\n\n    old_workloop = Worker.workloop\n\n    def sentry_workloop(*args, **kwargs):\n        try:\n            return old_workloop(*args, **kwargs)\n        finally:\n            with capture_internal_exceptions():\n                hub = Hub.current\n                if hub.get_integration(CeleryIntegration) is not None:\n                    hub.flush()\n\n    Worker.workloop = sentry_workloop\n", "path": "sentry_sdk/integrations/celery.py"}], "after_files": [{"content": "from __future__ import absolute_import\n\nimport sys\n\nfrom celery.exceptions import (  # type: ignore\n    SoftTimeLimitExceeded,\n    Retry,\n    Ignore,\n    Reject,\n)\n\nfrom sentry_sdk.hub import Hub\nfrom sentry_sdk.utils import capture_internal_exceptions, event_from_exception\nfrom sentry_sdk.tracing import SpanContext\nfrom sentry_sdk._compat import reraise\nfrom sentry_sdk.integrations import Integration\nfrom sentry_sdk.integrations.logging import ignore_logger\n\n\nCELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)\n\n\nclass CeleryIntegration(Integration):\n    identifier = \"celery\"\n\n    def __init__(self, propagate_traces=True):\n        self.propagate_traces = propagate_traces\n\n    @staticmethod\n    def setup_once():\n        import celery.app.trace as trace  # type: ignore\n\n        old_build_tracer = trace.build_tracer\n\n        def sentry_build_tracer(name, task, *args, **kwargs):\n            if not getattr(task, \"_sentry_is_patched\", False):\n                # Need to patch both methods because older celery sometimes\n                # short-circuits to task.run if it thinks it's safe.\n                task.__call__ = _wrap_task_call(task, task.__call__)\n                task.run = _wrap_task_call(task, task.run)\n                task.apply_async = _wrap_apply_async(task, task.apply_async)\n\n                # `build_tracer` is apparently called for every task\n                # invocation. Can't wrap every celery task for every invocation\n                # or we will get infinitely nested wrapper functions.\n                task._sentry_is_patched = True\n\n            return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))\n\n        trace.build_tracer = sentry_build_tracer\n\n        _patch_worker_exit()\n\n        # This logger logs every status of every task that ran on the worker.\n        # Meaning that every task's breadcrumbs are full of stuff like \"Task\n        # <foo> raised unexpected <bar>\".\n        ignore_logger(\"celery.worker.job\")\n\n\ndef _wrap_apply_async(task, f):\n    def apply_async(*args, **kwargs):\n        hub = Hub.current\n        integration = hub.get_integration(CeleryIntegration)\n        if integration is not None and integration.propagate_traces:\n            headers = None\n            for key, value in hub.iter_trace_propagation_headers():\n                if headers is None:\n                    headers = dict(kwargs.get(\"headers\") or {})\n                headers[key] = value\n            if headers is not None:\n                kwargs[\"headers\"] = headers\n        return f(*args, **kwargs)\n\n    return apply_async\n\n\ndef _wrap_tracer(task, f):\n    # Need to wrap tracer for pushing the scope before prerun is sent, and\n    # popping it after postrun is sent.\n    #\n    # This is the reason we don't use signals for hooking in the first place.\n    # Also because in Celery 3, signal dispatch returns early if one handler\n    # crashes.\n    def _inner(*args, **kwargs):\n        hub = Hub.current\n        if hub.get_integration(CeleryIntegration) is None:\n            return f(*args, **kwargs)\n\n        with hub.push_scope() as scope:\n            scope._name = \"celery\"\n            scope.clear_breadcrumbs()\n            _continue_trace(args[3].get(\"headers\") or {}, scope)\n            scope.add_event_processor(_make_event_processor(task, *args, **kwargs))\n\n            return f(*args, **kwargs)\n\n    return _inner\n\n\ndef _continue_trace(headers, scope):\n    if headers:\n        span_context = SpanContext.continue_from_headers(headers)\n    else:\n        span_context = SpanContext.start_trace()\n    scope.set_span_context(span_context)\n\n\ndef _wrap_task_call(task, f):\n    # Need to wrap task call because the exception is caught before we get to\n    # see it. Also celery's reported stacktrace is untrustworthy.\n    def _inner(*args, **kwargs):\n        try:\n            return f(*args, **kwargs)\n        except Exception:\n            exc_info = sys.exc_info()\n            with capture_internal_exceptions():\n                _capture_exception(task, exc_info)\n            reraise(*exc_info)\n\n    return _inner\n\n\ndef _make_event_processor(task, uuid, args, kwargs, request=None):\n    def event_processor(event, hint):\n        with capture_internal_exceptions():\n            event[\"transaction\"] = task.name\n\n        with capture_internal_exceptions():\n            extra = event.setdefault(\"extra\", {})\n            extra[\"celery-job\"] = {\n                \"task_name\": task.name,\n                \"args\": args,\n                \"kwargs\": kwargs,\n            }\n\n        if \"exc_info\" in hint:\n            with capture_internal_exceptions():\n                if issubclass(hint[\"exc_info\"][0], SoftTimeLimitExceeded):\n                    event[\"fingerprint\"] = [\n                        \"celery\",\n                        \"SoftTimeLimitExceeded\",\n                        getattr(task, \"name\", task),\n                    ]\n\n        return event\n\n    return event_processor\n\n\ndef _capture_exception(task, exc_info):\n    hub = Hub.current\n\n    if hub.get_integration(CeleryIntegration) is None:\n        return\n    if isinstance(exc_info[1], CELERY_CONTROL_FLOW_EXCEPTIONS):\n        return\n    if hasattr(task, \"throws\") and isinstance(exc_info[1], task.throws):\n        return\n\n    event, hint = event_from_exception(\n        exc_info,\n        client_options=hub.client.options,\n        mechanism={\"type\": \"celery\", \"handled\": False},\n    )\n\n    hub.capture_event(event, hint=hint)\n\n\ndef _patch_worker_exit():\n    # Need to flush queue before worker shutdown because a crashing worker will\n    # call os._exit\n    from billiard.pool import Worker  # type: ignore\n\n    old_workloop = Worker.workloop\n\n    def sentry_workloop(*args, **kwargs):\n        try:\n            return old_workloop(*args, **kwargs)\n        finally:\n            with capture_internal_exceptions():\n                hub = Hub.current\n                if hub.get_integration(CeleryIntegration) is not None:\n                    hub.flush()\n\n    Worker.workloop = sentry_workloop\n", "path": "sentry_sdk/integrations/celery.py"}]}