We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Unable to save pickled objects with couchbase as result backend
Hi it seems like when I attempt to process groups of chords, the couchbase result backend is consistently failing to unlock the chord when reading from the db:

`celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2] retry: Retry in 1s: ValueFormatError()`

This behavior does not occur with the redis result backend, i can switch between them and see that the error unlocking only occurs on couchbase.

## Steps to reproduce
Attempt to process a chord with couchbase backend using pickle serialization.

## Expected behavior
Chords process correctly, and resulting data is fed to the next task

## Actual behavior
Celery is unable to unlock the chord from the result backend

## Celery project info: 
```
celery -A ipaassteprunner report

software -> celery:4.1.0 (latentcall) kombu:4.1.0 py:2.7.10
            billiard:3.5.0.3 py-amqp:2.2.2
platform -> system:Darwin arch:64bit imp:CPython
loader   -> celery.loaders.app.AppLoader
settings -> transport:pyamqp results:couchbase://isadmin:**@localhost:8091/tasks

task_serializer: 'pickle'
result_serializer: 'pickle'
dbconfig: <ipaascommon.ipaas_config.DatabaseConfig object at 0x10fbbfe10>
db_pass: u'********'
IpaasConfig: <class 'ipaascommon.ipaas_config.IpaasConfig'>
imports:
    ('ipaassteprunner.tasks',)
worker_redirect_stdouts: False
DatabaseConfig: u'********'
db_port: '8091'
ipaas_constants: <module 'ipaascommon.ipaas_constants' from '/Library/Python/2.7/site-packages/ipaascommon/ipaas_constants.pyc'>
enable_utc: True
db_user: 'isadmin'
db_host: 'localhost'
result_backend: u'couchbase://isadmin:********@localhost:8091/tasks'
result_expires: 3600
iconfig: <ipaascommon.ipaas_config.IpaasConfig object at 0x10fbbfd90>
broker_url: u'amqp://guest:********@localhost:5672//'
task_bucket: 'tasks'
accept_content: ['pickle']
```
### Additional Debug output
```
[2017-12-13 15:39:57,860: INFO/MainProcess] Received task: celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2]  ETA:[2017-12-13 20:39:58.853535+00:00] 
[2017-12-13 15:39:57,861: DEBUG/MainProcess] basic.qos: prefetch_count->27
[2017-12-13 15:39:58,859: DEBUG/MainProcess] TaskPool: Apply <function _fast_trace_task at 0x10b410b90> (args:('celery.chord_unlock', 'e3139ae5-a67d-4f0c-8c54-73b1e19433d2', {'origin': 'gen53678@silo2460', 'lang': 'py', 'task': 'celery.chord_unlock', 'group': None, 'root_id': '0acd3e0d-7532-445c-8916-b5fc8a6395ab', u'delivery_info': {u'priority': None, u'redelivered': False, u'routing_key': u'celery', u'exchange': u''}, 'expires': None, u'correlation_id': 'e3139ae5-a67d-4f0c-8c54-73b1e19433d2', 'retries': 311, 'timelimit': [None, None], 'argsrepr': "('90c64bef-21ba-42f9-be75-fdd724375a7a', {'chord_size': 2, 'task': 'ipaassteprunner.tasks.transfer_data', 'subtask_type': None, 'kwargs': {}, 'args': (), 'options': {'chord_size': None, 'chain': [...], 'task_id': '9c6b5e1c-2089-4db7-9590-117aeaf782c7', 'root_id': '0acd3e0d-7532-445c-8916-b5fc8a6395ab', 'parent_id': 'c27c9565-19a6-4683-8180-60f0c25007e9', 'reply_to': '0a58093c-6fdd-3458-9a34-7d5e094ac6a8'}, 'immutable': False})", 'eta': '2017-12-13T20:39:58.853535+00:00', 'parent_id': 'c27c9565-19a6-4683-8180-60f0c25007e9', u'reply_to':... kwargs:{})
[2017-12-13 15:40:00,061: DEBUG/MainProcess] basic.qos: prefetch_count->26
[2017-12-13 15:40:00,065: DEBUG/MainProcess] Task accepted: celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2] pid:53679
[2017-12-13 15:40:00,076: INFO/ForkPoolWorker-6] Task celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2] retry: Retry in 1s: ValueFormatError()
```

### Stack trace from chord unlocking failure
```python
Traceback (most recent call last):
  File "/Library/Python/2.7/site-packages/celery/app/trace.py", line 374, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/Library/Python/2.7/site-packages/celery/app/trace.py", line 629, in __protected_call__
    return self.run(*args, **kwargs)
  File "/Library/Python/2.7/site-packages/celery/app/builtins.py", line 75, in unlock_chord
    raise self.retry(countdown=interval, max_retries=max_retries)
  File "/Library/Python/2.7/site-packages/celery/app/task.py", line 689, in retry
    raise ret
Retry: Retry in 1s
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/backends/couchbase.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """Couchbase result store backend."""
3 from __future__ import absolute_import, unicode_literals
4 
5 import logging
6 
7 from kombu.utils.encoding import str_t
8 from kombu.utils.url import _parse_url
9 
10 from celery.exceptions import ImproperlyConfigured
11 
12 from .base import KeyValueStoreBackend
13 
14 try:
15     import couchbase_ffi  # noqa
16 except ImportError:
17     pass  # noqa
18 try:
19     from couchbase import Couchbase
20     from couchbase.connection import Connection
21     from couchbase.exceptions import NotFoundError
22 except ImportError:
23     Couchbase = Connection = NotFoundError = None   # noqa
24 
25 __all__ = ('CouchbaseBackend',)
26 
27 
28 class CouchbaseBackend(KeyValueStoreBackend):
29     """Couchbase backend.
30 
31     Raises:
32         celery.exceptions.ImproperlyConfigured:
33             if module :pypi:`couchbase` is not available.
34     """
35 
36     bucket = 'default'
37     host = 'localhost'
38     port = 8091
39     username = None
40     password = None
41     quiet = False
42     supports_autoexpire = True
43 
44     timeout = 2.5
45 
46     # Use str as couchbase key not bytes
47     key_t = str_t
48 
49     def __init__(self, url=None, *args, **kwargs):
50         kwargs.setdefault('expires_type', int)
51         super(CouchbaseBackend, self).__init__(*args, **kwargs)
52         self.url = url
53 
54         if Couchbase is None:
55             raise ImproperlyConfigured(
56                 'You need to install the couchbase library to use the '
57                 'Couchbase backend.',
58             )
59 
60         uhost = uport = uname = upass = ubucket = None
61         if url:
62             _, uhost, uport, uname, upass, ubucket, _ = _parse_url(url)
63             ubucket = ubucket.strip('/') if ubucket else None
64 
65         config = self.app.conf.get('couchbase_backend_settings', None)
66         if config is not None:
67             if not isinstance(config, dict):
68                 raise ImproperlyConfigured(
69                     'Couchbase backend settings should be grouped in a dict',
70                 )
71         else:
72             config = {}
73 
74         self.host = uhost or config.get('host', self.host)
75         self.port = int(uport or config.get('port', self.port))
76         self.bucket = ubucket or config.get('bucket', self.bucket)
77         self.username = uname or config.get('username', self.username)
78         self.password = upass or config.get('password', self.password)
79 
80         self._connection = None
81 
82     def _get_connection(self):
83         """Connect to the Couchbase server."""
84         if self._connection is None:
85             kwargs = {'bucket': self.bucket, 'host': self.host}
86 
87             if self.port:
88                 kwargs.update({'port': self.port})
89             if self.username:
90                 kwargs.update({'username': self.username})
91             if self.password:
92                 kwargs.update({'password': self.password})
93 
94             logging.debug('couchbase settings %r', kwargs)
95             self._connection = Connection(**kwargs)
96         return self._connection
97 
98     @property
99     def connection(self):
100         return self._get_connection()
101 
102     def get(self, key):
103         try:
104             return self.connection.get(key).value
105         except NotFoundError:
106             return None
107 
108     def set(self, key, value):
109         self.connection.set(key, value, ttl=self.expires)
110 
111     def mget(self, keys):
112         return [self.get(key) for key in keys]
113 
114     def delete(self, key):
115         self.connection.delete(key)
116
```
--- 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/celery/backends/couchbase.py b/celery/backends/couchbase.py
--- a/celery/backends/couchbase.py
+++ b/celery/backends/couchbase.py
@@ -19,6 +19,7 @@
     from couchbase import Couchbase
     from couchbase.connection import Connection
     from couchbase.exceptions import NotFoundError
+    from couchbase import FMT_AUTO
 except ImportError:
     Couchbase = Connection = NotFoundError = None   # noqa
 
@@ -106,7 +107,7 @@
             return None
 
     def set(self, key, value):
-        self.connection.set(key, value, ttl=self.expires)
+        self.connection.set(key, value, ttl=self.expires, format=FMT_AUTO)
 
     def mget(self, keys):
         return [self.get(key) for key in keys]

{"golden_diff": "diff --git a/celery/backends/couchbase.py b/celery/backends/couchbase.py\n--- a/celery/backends/couchbase.py\n+++ b/celery/backends/couchbase.py\n@@ -19,6 +19,7 @@\n     from couchbase import Couchbase\n     from couchbase.connection import Connection\n     from couchbase.exceptions import NotFoundError\n+    from couchbase import FMT_AUTO\n except ImportError:\n     Couchbase = Connection = NotFoundError = None   # noqa\n \n@@ -106,7 +107,7 @@\n             return None\n \n     def set(self, key, value):\n-        self.connection.set(key, value, ttl=self.expires)\n+        self.connection.set(key, value, ttl=self.expires, format=FMT_AUTO)\n \n     def mget(self, keys):\n         return [self.get(key) for key in keys]\n", "issue": "Unable to save pickled objects with couchbase as result backend\nHi it seems like when I attempt to process groups of chords, the couchbase result backend is consistently failing to unlock the chord when reading from the db:\r\n\r\n`celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2] retry: Retry in 1s: ValueFormatError()`\r\n\r\nThis behavior does not occur with the redis result backend, i can switch between them and see that the error unlocking only occurs on couchbase.\r\n\r\n## Steps to reproduce\r\nAttempt to process a chord with couchbase backend using pickle serialization.\r\n\r\n## Expected behavior\r\nChords process correctly, and resulting data is fed to the next task\r\n\r\n## Actual behavior\r\nCelery is unable to unlock the chord from the result backend\r\n\r\n## Celery project info: \r\n```\r\ncelery -A ipaassteprunner report\r\n\r\nsoftware -> celery:4.1.0 (latentcall) kombu:4.1.0 py:2.7.10\r\n            billiard:3.5.0.3 py-amqp:2.2.2\r\nplatform -> system:Darwin arch:64bit imp:CPython\r\nloader   -> celery.loaders.app.AppLoader\r\nsettings -> transport:pyamqp results:couchbase://isadmin:**@localhost:8091/tasks\r\n\r\ntask_serializer: 'pickle'\r\nresult_serializer: 'pickle'\r\ndbconfig: <ipaascommon.ipaas_config.DatabaseConfig object at 0x10fbbfe10>\r\ndb_pass: u'********'\r\nIpaasConfig: <class 'ipaascommon.ipaas_config.IpaasConfig'>\r\nimports:\r\n    ('ipaassteprunner.tasks',)\r\nworker_redirect_stdouts: False\r\nDatabaseConfig: u'********'\r\ndb_port: '8091'\r\nipaas_constants: <module 'ipaascommon.ipaas_constants' from '/Library/Python/2.7/site-packages/ipaascommon/ipaas_constants.pyc'>\r\nenable_utc: True\r\ndb_user: 'isadmin'\r\ndb_host: 'localhost'\r\nresult_backend: u'couchbase://isadmin:********@localhost:8091/tasks'\r\nresult_expires: 3600\r\niconfig: <ipaascommon.ipaas_config.IpaasConfig object at 0x10fbbfd90>\r\nbroker_url: u'amqp://guest:********@localhost:5672//'\r\ntask_bucket: 'tasks'\r\naccept_content: ['pickle']\r\n```\r\n### Additional Debug output\r\n```\r\n[2017-12-13 15:39:57,860: INFO/MainProcess] Received task: celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2]  ETA:[2017-12-13 20:39:58.853535+00:00] \r\n[2017-12-13 15:39:57,861: DEBUG/MainProcess] basic.qos: prefetch_count->27\r\n[2017-12-13 15:39:58,859: DEBUG/MainProcess] TaskPool: Apply <function _fast_trace_task at 0x10b410b90> (args:('celery.chord_unlock', 'e3139ae5-a67d-4f0c-8c54-73b1e19433d2', {'origin': 'gen53678@silo2460', 'lang': 'py', 'task': 'celery.chord_unlock', 'group': None, 'root_id': '0acd3e0d-7532-445c-8916-b5fc8a6395ab', u'delivery_info': {u'priority': None, u'redelivered': False, u'routing_key': u'celery', u'exchange': u''}, 'expires': None, u'correlation_id': 'e3139ae5-a67d-4f0c-8c54-73b1e19433d2', 'retries': 311, 'timelimit': [None, None], 'argsrepr': \"('90c64bef-21ba-42f9-be75-fdd724375a7a', {'chord_size': 2, 'task': 'ipaassteprunner.tasks.transfer_data', 'subtask_type': None, 'kwargs': {}, 'args': (), 'options': {'chord_size': None, 'chain': [...], 'task_id': '9c6b5e1c-2089-4db7-9590-117aeaf782c7', 'root_id': '0acd3e0d-7532-445c-8916-b5fc8a6395ab', 'parent_id': 'c27c9565-19a6-4683-8180-60f0c25007e9', 'reply_to': '0a58093c-6fdd-3458-9a34-7d5e094ac6a8'}, 'immutable': False})\", 'eta': '2017-12-13T20:39:58.853535+00:00', 'parent_id': 'c27c9565-19a6-4683-8180-60f0c25007e9', u'reply_to':... kwargs:{})\r\n[2017-12-13 15:40:00,061: DEBUG/MainProcess] basic.qos: prefetch_count->26\r\n[2017-12-13 15:40:00,065: DEBUG/MainProcess] Task accepted: celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2] pid:53679\r\n[2017-12-13 15:40:00,076: INFO/ForkPoolWorker-6] Task celery.chord_unlock[e3139ae5-a67d-4f0c-8c54-73b1e19433d2] retry: Retry in 1s: ValueFormatError()\r\n```\r\n\r\n### Stack trace from chord unlocking failure\r\n```python\r\nTraceback (most recent call last):\r\n  File \"/Library/Python/2.7/site-packages/celery/app/trace.py\", line 374, in trace_task\r\n    R = retval = fun(*args, **kwargs)\r\n  File \"/Library/Python/2.7/site-packages/celery/app/trace.py\", line 629, in __protected_call__\r\n    return self.run(*args, **kwargs)\r\n  File \"/Library/Python/2.7/site-packages/celery/app/builtins.py\", line 75, in unlock_chord\r\n    raise self.retry(countdown=interval, max_retries=max_retries)\r\n  File \"/Library/Python/2.7/site-packages/celery/app/task.py\", line 689, in retry\r\n    raise ret\r\nRetry: Retry in 1s\r\n```\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"Couchbase result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport logging\n\nfrom kombu.utils.encoding import str_t\nfrom kombu.utils.url import _parse_url\n\nfrom celery.exceptions import ImproperlyConfigured\n\nfrom .base import KeyValueStoreBackend\n\ntry:\n    import couchbase_ffi  # noqa\nexcept ImportError:\n    pass  # noqa\ntry:\n    from couchbase import Couchbase\n    from couchbase.connection import Connection\n    from couchbase.exceptions import NotFoundError\nexcept ImportError:\n    Couchbase = Connection = NotFoundError = None   # noqa\n\n__all__ = ('CouchbaseBackend',)\n\n\nclass CouchbaseBackend(KeyValueStoreBackend):\n    \"\"\"Couchbase backend.\n\n    Raises:\n        celery.exceptions.ImproperlyConfigured:\n            if module :pypi:`couchbase` is not available.\n    \"\"\"\n\n    bucket = 'default'\n    host = 'localhost'\n    port = 8091\n    username = None\n    password = None\n    quiet = False\n    supports_autoexpire = True\n\n    timeout = 2.5\n\n    # Use str as couchbase key not bytes\n    key_t = str_t\n\n    def __init__(self, url=None, *args, **kwargs):\n        kwargs.setdefault('expires_type', int)\n        super(CouchbaseBackend, self).__init__(*args, **kwargs)\n        self.url = url\n\n        if Couchbase is None:\n            raise ImproperlyConfigured(\n                'You need to install the couchbase library to use the '\n                'Couchbase backend.',\n            )\n\n        uhost = uport = uname = upass = ubucket = None\n        if url:\n            _, uhost, uport, uname, upass, ubucket, _ = _parse_url(url)\n            ubucket = ubucket.strip('/') if ubucket else None\n\n        config = self.app.conf.get('couchbase_backend_settings', None)\n        if config is not None:\n            if not isinstance(config, dict):\n                raise ImproperlyConfigured(\n                    'Couchbase backend settings should be grouped in a dict',\n                )\n        else:\n            config = {}\n\n        self.host = uhost or config.get('host', self.host)\n        self.port = int(uport or config.get('port', self.port))\n        self.bucket = ubucket or config.get('bucket', self.bucket)\n        self.username = uname or config.get('username', self.username)\n        self.password = upass or config.get('password', self.password)\n\n        self._connection = None\n\n    def _get_connection(self):\n        \"\"\"Connect to the Couchbase server.\"\"\"\n        if self._connection is None:\n            kwargs = {'bucket': self.bucket, 'host': self.host}\n\n            if self.port:\n                kwargs.update({'port': self.port})\n            if self.username:\n                kwargs.update({'username': self.username})\n            if self.password:\n                kwargs.update({'password': self.password})\n\n            logging.debug('couchbase settings %r', kwargs)\n            self._connection = Connection(**kwargs)\n        return self._connection\n\n    @property\n    def connection(self):\n        return self._get_connection()\n\n    def get(self, key):\n        try:\n            return self.connection.get(key).value\n        except NotFoundError:\n            return None\n\n    def set(self, key, value):\n        self.connection.set(key, value, ttl=self.expires)\n\n    def mget(self, keys):\n        return [self.get(key) for key in keys]\n\n    def delete(self, key):\n        self.connection.delete(key)\n", "path": "celery/backends/couchbase.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"Couchbase result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport logging\n\nfrom kombu.utils.encoding import str_t\nfrom kombu.utils.url import _parse_url\n\nfrom celery.exceptions import ImproperlyConfigured\n\nfrom .base import KeyValueStoreBackend\n\ntry:\n    import couchbase_ffi  # noqa\nexcept ImportError:\n    pass  # noqa\ntry:\n    from couchbase import Couchbase\n    from couchbase.connection import Connection\n    from couchbase.exceptions import NotFoundError\n    from couchbase import FMT_AUTO\nexcept ImportError:\n    Couchbase = Connection = NotFoundError = None   # noqa\n\n__all__ = ('CouchbaseBackend',)\n\n\nclass CouchbaseBackend(KeyValueStoreBackend):\n    \"\"\"Couchbase backend.\n\n    Raises:\n        celery.exceptions.ImproperlyConfigured:\n            if module :pypi:`couchbase` is not available.\n    \"\"\"\n\n    bucket = 'default'\n    host = 'localhost'\n    port = 8091\n    username = None\n    password = None\n    quiet = False\n    supports_autoexpire = True\n\n    timeout = 2.5\n\n    # Use str as couchbase key not bytes\n    key_t = str_t\n\n    def __init__(self, url=None, *args, **kwargs):\n        kwargs.setdefault('expires_type', int)\n        super(CouchbaseBackend, self).__init__(*args, **kwargs)\n        self.url = url\n\n        if Couchbase is None:\n            raise ImproperlyConfigured(\n                'You need to install the couchbase library to use the '\n                'Couchbase backend.',\n            )\n\n        uhost = uport = uname = upass = ubucket = None\n        if url:\n            _, uhost, uport, uname, upass, ubucket, _ = _parse_url(url)\n            ubucket = ubucket.strip('/') if ubucket else None\n\n        config = self.app.conf.get('couchbase_backend_settings', None)\n        if config is not None:\n            if not isinstance(config, dict):\n                raise ImproperlyConfigured(\n                    'Couchbase backend settings should be grouped in a dict',\n                )\n        else:\n            config = {}\n\n        self.host = uhost or config.get('host', self.host)\n        self.port = int(uport or config.get('port', self.port))\n        self.bucket = ubucket or config.get('bucket', self.bucket)\n        self.username = uname or config.get('username', self.username)\n        self.password = upass or config.get('password', self.password)\n\n        self._connection = None\n\n    def _get_connection(self):\n        \"\"\"Connect to the Couchbase server.\"\"\"\n        if self._connection is None:\n            kwargs = {'bucket': self.bucket, 'host': self.host}\n\n            if self.port:\n                kwargs.update({'port': self.port})\n            if self.username:\n                kwargs.update({'username': self.username})\n            if self.password:\n                kwargs.update({'password': self.password})\n\n            logging.debug('couchbase settings %r', kwargs)\n            self._connection = Connection(**kwargs)\n        return self._connection\n\n    @property\n    def connection(self):\n        return self._get_connection()\n\n    def get(self, key):\n        try:\n            return self.connection.get(key).value\n        except NotFoundError:\n            return None\n\n    def set(self, key, value):\n        self.connection.set(key, value, ttl=self.expires, format=FMT_AUTO)\n\n    def mget(self, keys):\n        return [self.get(key) for key in keys]\n\n    def delete(self, key):\n        self.connection.delete(key)\n", "path": "celery/backends/couchbase.py"}]}