store: implement delta uploads in push.

debian/control

@@ -19,6 +19,7 @@ Build-Depends: bash-completion,
                python3-pkg-resources,
                python3-progressbar,
                python3-pymacaroons,
+               python3-pysha3,
                python3-requests,
                python3-requests-toolbelt,
                python3-responses,
@@ -49,6 +50,7 @@ Depends: python3-apt,
          python3-pkg-resources,
          python3-progressbar,
          python3-pymacaroons,
+         python3-pysha3,
          python3-requests,
          python3-requests-toolbelt,
          python3-simplejson,

requirements.txt

@@ -14,6 +14,7 @@ responses==0.5.1
 petname==1.12
 pymacaroons==0.9.2
 pymacaroons-pynacl==0.9.3
+pysha3==1.0b1
 simplejson==3.8.2
 tabulate==0.7.5
 python-debian==0.1.28

setup.py

@@ -56,6 +56,7 @@
             ['libraries/' + x for x in os.listdir('libraries')]),
     ],
     install_requires=[
+        'pysha3',
         'pyxdg',
         'requests',
         'libarchive-c',

snapcraft/_store.py

@@ -17,6 +17,7 @@
 from contextlib import contextmanager
 import datetime
 import getpass
+import hashlib
 import json
 import logging
 import operator
@@ -30,11 +31,19 @@
 from tabulate import tabulate
 import yaml
 
+from snapcraft.file_utils import calculate_sha3_384
 from snapcraft import storeapi
+from snapcraft.storeapi.errors import StoreDeltaApplicationError
 from snapcraft.internal import (
     cache,
+    deltas,
     repo,
 )
+from snapcraft.internal.deltas.errors import (
+    DeltaGenerationError,
+    DeltaGenerationTooBigError,
+    DeltaToolError,
+)
 
 
 logger = logging.getLogger(__name__)
@@ -402,6 +411,12 @@ def sign_build(snap_filename, key_name=None, local=False):
 def push(snap_filename, release_channels=None):
     """Push a snap_filename to the store.
 
+    If the DELTA_UPLOADS_EXPERIMENTAL environment variable is set
+    and a cached snap is available, a delta will be generated from
+    the cached snap to the new target snap and uploaded instead. In the
+    case of a delta processing or upload failure, push will fall back to
+    uploading the full snap.
+
     If release_channels is defined it also releases it to those channels if the
     store deems the uploaded snap as ready to release.
     """
@@ -417,27 +432,100 @@ def push(snap_filename, release_channels=None):
     with _requires_login():
         store.push_precheck(snap_name)
 
-    with _requires_login():
-        tracker = store.upload(snap_name, snap_filename)
+    snap_cache = cache.SnapCache(project_name=snap_name)
+    arch = snap_yaml['architectures'][0]
+    source_snap = snap_cache.get(deb_arch=arch)
+
+    sha3_384_available = hasattr(hashlib, 'sha3_384')
+
+    if (os.environ.get('DELTA_UPLOADS_EXPERIMENTAL') and
+            sha3_384_available and source_snap):
+        try:
+            result = _push_delta(snap_name, snap_filename, source_snap)
+        except StoreDeltaApplicationError as e:
+            logger.warning(
+                'Error generating delta: {}\n'
+                'Falling back to pushing full snap...'.format(str(e)))
+            result = _push_snap(snap_name, snap_filename)
+        except storeapi.errors.StorePushError as e:
+            store_error = e.error_list[0].get('message')
+            logger.warning(
+                'Unable to push delta to store: {}\n'
+                'Falling back to pushing full snap...'.format(store_error))
+            result = _push_snap(snap_name, snap_filename)
+    else:
+        result = _push_snap(snap_name, snap_filename)
 
-    result = tracker.track()
     # This is workaround until LP: #1599875 is solved
     if 'revision' in result:
         logger.info('Revision {!r} of {!r} created.'.format(
             result['revision'], snap_name))
-    else:
-        logger.info('Uploaded {!r}'.format(snap_name))
-    tracker.raise_for_code()
 
-    if os.environ.get('DELTA_UPLOADS_EXPERIMENTAL'):
-        snap_cache = cache.SnapCache(project_name=snap_name)
-        snap_cache.cache(snap_filename, result['revision'])
-        snap_cache.prune(keep_revision=result['revision'])
+        if os.environ.get('DELTA_UPLOADS_EXPERIMENTAL'):
+            snap_cache.cache(snap_filename=snap_filename)
+            snap_cache.prune(deb_arch=arch,
+                             keep_hash=calculate_sha3_384(snap_filename))
+    else:
+        logger.info('Pushing {!r}'.format(snap_name))
 
     if release_channels:
         release(snap_name, result['revision'], release_channels)
 
 
+def _push_snap(snap_name, snap_filename):
+    store = storeapi.StoreClient()
+    with _requires_login():
+        tracker = store.upload(snap_name, snap_filename)
+    result = tracker.track()
+    tracker.raise_for_code()
+    return result
+
+
+def _push_delta(snap_name, snap_filename, source_snap):
+    store = storeapi.StoreClient()
+    delta_format = 'xdelta3'
+    logger.info('Found cached source snap {}.'.format(source_snap))
+    target_snap = os.path.join(os.getcwd(), snap_filename)
+
+    try:
+        xdelta_generator = deltas.XDelta3Generator(
+            source_path=source_snap, target_path=target_snap)
+        delta_filename = xdelta_generator.make_delta()
+    except (DeltaGenerationError, DeltaGenerationTooBigError,
+            DeltaToolError) as e:
+        raise StoreDeltaApplicationError(str(e))
+
+    snap_hashes = {'source_hash': calculate_sha3_384(source_snap),
+                   'target_hash': calculate_sha3_384(target_snap),
+                   'delta_hash': calculate_sha3_384(delta_filename)}
+
+    try:
+        logger.info('Pushing delta {}.'.format(delta_filename))
+        with _requires_login():
+            delta_tracker = store.upload(
+                snap_name,
+                delta_filename,
+                delta_format=delta_format,
+                source_hash=snap_hashes['source_hash'],
+                target_hash=snap_hashes['target_hash'],
+                delta_hash=snap_hashes['delta_hash'])
+        result = delta_tracker.track()
+        delta_tracker.raise_for_code()
+    except storeapi.errors.StoreReviewError as e:
+        if e.code == 'processing_upload_delta_error':
+            raise StoreDeltaApplicationError
+        else:
+            raise
+    finally:
+        if os.path.isfile(delta_filename):
+            try:
+                os.remove(delta_filename)
+            except OSError:
+                logger.warning(
+                    'Unable to remove delta {}.'.format(delta_filename))
+    return result
+
+
 def _get_text_for_opened_channels(opened_channels):
     if len(opened_channels) == 1:
         return 'The {!r} channel is now open.'.format(opened_channels[0])

snapcraft/file_utils.py

@@ -15,17 +15,23 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from contextlib import contextmanager
+import hashlib
+import logging
 import os
 import shutil
 import subprocess
-import logging
+import sys
+
 
 from snapcraft.internal.errors import (
     RequiredCommandFailure,
     RequiredCommandNotFound,
     RequiredPathDoesNotExist,
 )
 
+if sys.version_info < (3, 6):
+    import sha3  # noqa
+
 
 logger = logging.getLogger(__name__)
 
@@ -209,3 +215,16 @@ def requires_path_exists(path, error_fmt=None):
             kwargs['fmt'] = error_fmt
         raise RequiredPathDoesNotExist(**kwargs)
     yield
+
+
+def calculate_sha3_384(path):
+    """Calculate sha3 384 hash, reading the file in 1MB chunks."""
+    blocksize = 2**20
+    with open(path, 'rb') as snap_file:
+        hasher = hashlib.sha3_384()
+        while True:
+            buf = snap_file.read(blocksize)
+            if not buf:
+                break
+            hasher.update(buf)
+        return hasher.hexdigest()

snapcraft/internal/cache/_cache.py

@@ -31,6 +31,9 @@ def __init__(self):
     def cache(self):
         raise NotImplementedError
 
+    def get(self):
+        raise NotImplementedError
+
     def prune(self, *args, **kwargs):
         raise NotImplementedError
 
@@ -40,4 +43,4 @@ class SnapcraftProjectCache(SnapcraftCache):
     def __init__(self, *, project_name):
         super().__init__()
         self.project_cache_root = os.path.join(
-            self.cache_root, project_name)
+            self.cache_root, 'projects', project_name)