[feature] retry during upload + proper wait

Added retry to upload and modified retry implementation to comply with
discussed spec.
According to it, we should wait between retries, something like 1s, 10s,
.. up to 1 minute.

-- Resolves: #8822

2 files changed, 45 insertions, 12 deletions

diff --git a/src/leap/soledad/client/_db/blobs.py b/src/leap/soledad/client/_db/blobs.py
index 05802dab..d25a5ca6 100644
--- a/src/leap/soledad/client/_db/blobs.py
+++ b/src/leap/soledad/client/_db/blobs.py
@@ -31,6 +31,7 @@ from functools import partial
 from twisted.logger import Logger
 from twisted.enterprise import adbapi
 from twisted.internet import defer
+from twisted.internet import reactor
 from twisted.internet import error
 
 import treq
@@ -145,6 +146,24 @@ def check_http_status(code, blob_id=None, flags=None):
         raise SoledadError("Server Error: %s" % code)
 
 
+def sleep(seconds):
+    d = defer.Deferred()
+    reactor.callLater(seconds, d.callback, None)
+    return d
+
+
+@defer.inlineCallbacks
+def with_retry(func, *args, **kwargs):
+    retry_wait, max_wait = 1, 60
+    while True:
+        try:
+            yield func(*args, **kwargs)
+            break
+        except(error.ConnectError, error.ConnectionClosed):
+            yield sleep(retry_wait)
+            retry_wait = min(retry_wait + 10, max_wait)
+
+
 class DecrypterBuffer(object):
 
     def __init__(self, blob_id, secret, tag):
@@ -302,7 +321,8 @@ class BlobManager(object):
             deferreds = []
             for _ in range(min(self.concurrency_limit, len(missing))):
                 blob_id = missing.pop()
-                deferreds.append(self.__send_one(blob_id, namespace))
+                d = with_retry(self.__send_one, blob_id, namespace)
+                deferreds.append(d)
             yield defer.gatherResults(deferreds)
 
     @defer.inlineCallbacks
@@ -339,21 +359,11 @@ class BlobManager(object):
             for _ in range(min(self.concurrency_limit, len(docs_we_want))):
                 blob_id = docs_we_want.pop()
                 logger.info("Fetching new doc: %s" % blob_id)
-                d = self.__with_retry(self.get, blob_id, namespace)
+                d = with_retry(self.get, blob_id, namespace)
                 deferreds.append(d)
             yield defer.gatherResults(deferreds)
 
     @defer.inlineCallbacks
-    def __with_retry(self, func, *args, **kwargs):
-        retries, max_retries = 0, 300
-        while retries < max_retries:
-            try:
-                yield func(*args, **kwargs)
-                break
-            except(error.ConnectError, error.ConnectionClosed):
-                retries += 1
-
-    @defer.inlineCallbacks
     def sync(self, namespace=''):
         yield self.refresh_sync_status_from_server(namespace)
         yield self.fetch_missing(namespace)
diff --git a/tests/server/test_blobs_server.py b/tests/server/test_blobs_server.py
index f27fa985..e3bc761d 100644
--- a/tests/server/test_blobs_server.py
+++ b/tests/server/test_blobs_server.py
@@ -257,6 +257,29 @@ class BlobServerTestCase(unittest.TestCase):
 
     @defer.inlineCallbacks
     @pytest.mark.usefixtures("method_tmpdir")
+    def test_send_missing_retry(self):
+        manager = BlobManager(self.tempdir, self.uri, self.secret,
+                              self.secret, uuid4().hex)
+        self.addCleanup(manager.close)
+        blob_id = 'remote_only_blob_id'
+        yield manager.local.put(blob_id, BytesIO("X"), size=1)
+        yield self.port.stopListening()
+
+        def sleep(x):
+            d = defer.Deferred()
+            reactor.callLater(x, d.callback, None)
+            return d
+        d = manager.send_missing()
+        yield sleep(1)
+        self.port = reactor.listenTCP(
+            self.host.port, self.site, interface='127.0.0.1')
+        yield d
+        result = yield manager._download_and_decrypt(blob_id)
+        self.assertIsNotNone(result)
+        self.assertEquals(result[0].getvalue(), "X")
+
+    @defer.inlineCallbacks
+    @pytest.mark.usefixtures("method_tmpdir")
     def test_sync_fetch_missing(self):
         manager = BlobManager(self.tempdir, self.uri, self.secret,
                               self.secret, uuid4().hex)