From 2772d76d0d66d9e094a435e110fd9f945d9e105c Mon Sep 17 00:00:00 2001
From: drebs <drebs@riseup.net>
Date: Thu, 5 Oct 2017 15:24:46 -0300
Subject: [bug] fix retries for blobs download

- add a MaximumRetriesError exception to encapsulate other exceptions.
- record the pending status before trying to download
- modify update_sync_status to insert or update
- modify retry tests to check number of retries
- add a test for download retry limit
---
 src/leap/soledad/client/_db/blobs.py | 63 ++++++++++++++++++++++++------------
 1 file changed, 43 insertions(+), 20 deletions(-)

(limited to 'src/leap/soledad/client/_db/blobs.py')

diff --git a/src/leap/soledad/client/_db/blobs.py b/src/leap/soledad/client/_db/blobs.py
index 527a6418..4edb77f4 100644
--- a/src/leap/soledad/client/_db/blobs.py
+++ b/src/leap/soledad/client/_db/blobs.py
@@ -151,6 +151,10 @@ class RetriableTransferError(Exception):
     pass
 
 
+class MaximumRetriesError(Exception):
+    pass
+
+
 def sleep(seconds):
     d = defer.Deferred()
     reactor.callLater(seconds, d.callback, None)
@@ -346,19 +350,21 @@ class BlobManager(object):
 
     @defer.inlineCallbacks
     def __send_one(self, blob_id, namespace, i, total):
-            logger.info("Sending blob to server (%d/%d): %s"
-                        % (i, total, blob_id))
-            fd = yield self.local.get(blob_id, namespace=namespace)
-            try:
-                yield self._encrypt_and_upload(blob_id, fd)
-                yield self.local.update_sync_status(blob_id, SyncStatus.SYNCED)
-            except Exception as e:
-                yield self.local.increment_retries(blob_id)
-                _, retries = yield self.local.get_sync_status(blob_id)
-                if retries > self.max_retries:
-                    failed_upload = SyncStatus.FAILED_UPLOAD
-                    yield self.local.update_sync_status(blob_id, failed_upload)
-                raise e
+        logger.info("Sending blob to server (%d/%d): %s"
+                    % (i, total, blob_id))
+        fd = yield self.local.get(blob_id, namespace=namespace)
+        try:
+            yield self._encrypt_and_upload(blob_id, fd)
+            yield self.local.update_sync_status(blob_id, SyncStatus.SYNCED)
+        except Exception as e:
+            yield self.local.increment_retries(blob_id)
+            res = yield self.local.get_sync_status(blob_id)
+            _, retries = res
+            if (retries + 1) > self.max_retries:
+                failed_upload = SyncStatus.FAILED_UPLOAD
+                yield self.local.update_sync_status(blob_id, failed_upload)
+                raise MaximumRetriesError(e)
+            raise e
 
     @defer.inlineCallbacks
     def fetch_missing(self, namespace=''):
@@ -489,6 +495,9 @@ class BlobManager(object):
             logger.info("Found blob in local database: %s" % blob_id)
             defer.returnValue(local_blob)
 
+        yield self.local.update_sync_status(
+            blob_id, SyncStatus.PENDING_DOWNLOAD, namespace=namespace)
+
         try:
             result = yield self._download_and_decrypt(blob_id, namespace)
         except Exception as e:
@@ -505,12 +514,14 @@ class BlobManager(object):
                 logger.error(message)
 
             yield self.local.increment_retries(blob_id)
+
             if (retries + 1) >= self.max_retries:
                 failed_download = SyncStatus.FAILED_DOWNLOAD
-                yield self.local.update_sync_status(blob_id, failed_download)
-                raise e
-            else:
-                raise RetriableTransferError(e)
+                yield self.local.update_sync_status(
+                    blob_id, failed_download, namespace=namespace)
+                raise MaximumRetriesError(e)
+
+            raise RetriableTransferError(e)
 
         if not result:
             defer.returnValue(None)
@@ -684,10 +695,22 @@ class SQLiteBlobBackend(object):
         else:
             defer.returnValue([])
 
-    def update_sync_status(self, blob_id, sync_status):
-        query = 'update blobs set sync_status = ? where blob_id = ?'
+    @defer.inlineCallbacks
+    def update_sync_status(self, blob_id, sync_status, namespace=""):
+        query = 'SELECT sync_status FROM blobs WHERE blob_id = ?'
+        result = yield self.dbpool.runQuery(query, (blob_id,))
+
+        if not result:
+            insert = 'INSERT INTO blobs'
+            insert += ' (blob_id, namespace, payload, sync_status)'
+            insert += ' VALUES (?, ?, zeroblob(0), ?)'
+            values = (blob_id, namespace, sync_status)
+            yield self.dbpool.runOperation(insert, values)
+            return
+
+        update = 'UPDATE blobs SET sync_status = ? WHERE blob_id = ?'
         values = (sync_status, blob_id,)
-        return self.dbpool.runQuery(query, values)
+        result = yield self.dbpool.runOperation(update, values)
 
     def update_batch_sync_status(self, blob_id_list, sync_status,
                                  namespace=''):
-- 
cgit v1.2.3