[gentoo-portage-dev] [PATCH 3/4] rsync: split out repo storage framework

[Zac Medico]

Since there aremany ways to manage repository storage, split out a repo
storage framework. The HardlinkQuarantineRepoStorage class implements
the existing default behavior, and the InplaceRepoStorage class
implements the legacy behavior (when sync-allow-hardlinks is disabled in
repos.conf).

Each class implements RepoStorageInterface, which uses coroutine methods
since coroutines are well-suited to the I/O bound tasks that these
methods perform. The _sync_decorator is used to convert coroutine
methods to synchronous methods, for smooth integration into the
surrounding synchronous code.

Bug: https://bugs.gentoo.org/662070
---
 lib/portage/repository/storage/__init__.py         |  0
 .../repository/storage/hardlink_quarantine.py      | 95 ++++++++++++++++++++++
 lib/portage/repository/storage/inplace.py          | 49 +++++++++++
 lib/portage/repository/storage/interface.py        | 87 ++++++++++++++++++++
 lib/portage/sync/controller.py                     |  1 +
 lib/portage/sync/modules/rsync/rsync.py            | 85 +++++--------------
 lib/portage/sync/syncbase.py                       | 31 +++++++
 7 files changed, 284 insertions(+), 64 deletions(-)
 create mode 100644 lib/portage/repository/storage/__init__.py
 create mode 100644 lib/portage/repository/storage/hardlink_quarantine.py
 create mode 100644 lib/portage/repository/storage/inplace.py
 create mode 100644 lib/portage/repository/storage/interface.py

diff --git a/lib/portage/repository/storage/__init__.py b/lib/portage/repository/storage/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/lib/portage/repository/storage/hardlink_quarantine.py b/lib/portage/repository/storage/hardlink_quarantine.py
new file mode 100644
index 000000000..7e9cf4493
--- /dev/null
+++ b/lib/portage/repository/storage/hardlink_quarantine.py
@@ -0,0 +1,95 @@
+# Copyright 2018 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+from portage import os
+from portage.repository.storage.interface import (
+	RepoStorageException,
+	RepoStorageInterface,
+)
+from portage.util.futures import asyncio
+from portage.util.futures.compat_coroutine import (
+	coroutine,
+	coroutine_return,
+)
+
+from _emerge.SpawnProcess import SpawnProcess
+
+
+class HardlinkQuarantineRepoStorage(RepoStorageInterface):
+	"""
+	This is the default storage module, since its quite compatible with
+	most configurations.
+
+	It's desirable to be able to create shared hardlinks between the
+	download directory and the normal repository, and this is facilitated
+	by making the download directory be a subdirectory of the normal
+	repository location (ensuring that no mountpoints are crossed).
+	Shared hardlinks are created by using the rsync --link-dest option.
+
+	Since the download is initially unverified, it is safest to save
+	it in a quarantine directory. The quarantine directory is also
+	useful for making the repository update more atomic, so that it
+	less likely that normal repository location will be observed in
+	a partially synced state.
+	"""
+	def __init__(self, repo, spawn_kwargs):
+		self._user_location = repo.location
+		self._update_location = None
+		self._spawn_kwargs = spawn_kwargs
+		self._current_update = None
+
+	@coroutine
+	def _check_call(self, cmd):
+		"""
+		Run cmd and raise RepoStorageException on failure.
+
+		@param cmd: command to executre
+		@type cmd: list
+		"""
+		p = SpawnProcess(args=cmd, scheduler=asyncio._wrap_loop(), **self._spawn_kwargs)
+		p.start()
+		if (yield p.async_wait()) != os.EX_OK:
+			raise RepoStorageException('command exited with status {}: {}'.\
+				format(p.returncode, ' '.join(cmd)))
+
+	@coroutine
+	def init_update(self):
+		update_location = os.path.join(self._user_location, '.tmp-unverified-download-quarantine')
+		yield self._check_call(['rm', '-rf', update_location])
+
+		# Use  rsync --link-dest to hardlink a files into self._update_location,
+		# since cp -l is not portable.
+		yield self._check_call(['rsync', '-a', '--link-dest', self._user_location,
+			'--exclude', '/{}'.format(os.path.basename(update_location)),
+			self._user_location + '/', update_location + '/'])
+
+		self._update_location = update_location
+
+		coroutine_return(self._update_location)
+
+	@property
+	def current_update(self):
+		if self._update_location is None:
+			raise RepoStorageException('current update does not exist')
+		return self._update_location
+
+	@coroutine
+	def commit_update(self):
+		update_location = self.current_update
+		self._update_location = None
+		yield self._check_call(['rsync', '-a', '--delete',
+			'--exclude', '/{}'.format(os.path.basename(update_location)),
+			update_location + '/', self._user_location + '/'])
+
+		yield self._check_call(['rm', '-rf', update_location])
+
+	@coroutine
+	def abort_update(self):
+		if self._update_location is not None:
+			update_location = self._update_location
+			self._update_location = None
+			yield self._check_call(['rm', '-rf', update_location])
+
+	@coroutine
+	def garbage_collection(self):
+		yield self.abort_update()
diff --git a/lib/portage/repository/storage/inplace.py b/lib/portage/repository/storage/inplace.py
new file mode 100644
index 000000000..f1117ad03
--- /dev/null
+++ b/lib/portage/repository/storage/inplace.py
@@ -0,0 +1,49 @@
+# Copyright 2018 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+from portage.repository.storage.interface import (
+	RepoStorageException,
+	RepoStorageInterface,
+)
+from portage.util.futures.compat_coroutine import coroutine, coroutine_return
+
+
+class InplaceRepoStorage(RepoStorageInterface):
+	"""
+	Legacy repo storage behavior, where updates are applied in-place.
+	This module is not recommended, since the repository is left in an
+	unspecified (possibly malicious) state if the update fails.
+	"""
+	def __init__(self, repo, spawn_kwargs):
+		self._user_location = repo.location
+		self._update_location = None
+
+	@coroutine
+	def init_update(self):
+		self._update_location = self._user_location
+		coroutine_return(self._update_location)
+		yield None
+
+	@property
+	def current_update(self):
+		if self._update_location is None:
+			raise RepoStorageException('current update does not exist')
+		return self._update_location
+
+	@coroutine
+	def commit_update(self):
+		self.current_update
+		self._update_location = None
+		coroutine_return()
+		yield None
+
+	@coroutine
+	def abort_update(self):
+		self._update_location = None
+		coroutine_return()
+		yield None
+
+	@coroutine
+	def garbage_collection(self):
+		coroutine_return()
+		yield None
diff --git a/lib/portage/repository/storage/interface.py b/lib/portage/repository/storage/interface.py
new file mode 100644
index 000000000..f83c42b84
--- /dev/null
+++ b/lib/portage/repository/storage/interface.py
@@ -0,0 +1,87 @@
+# Copyright 2018 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+from portage.exception import PortageException
+from portage.util.futures.compat_coroutine import coroutine
+
+
+class RepoStorageException(PortageException):
+	"""
+	Base class for exceptions raise by RepoStorageInterface.
+	"""
+
+
+class RepoStorageInterface(object):
+	"""
+	Abstract repository storage interface.
+
+	Implementations can assume that the repo.location directory already
+	exists with appropriate permissions (SyncManager handles this).
+
+	TODO: Add a method to check of a previous uncommitted update, which
+	typically indicates a verification failure:
+	    https://bugs.gentoo.org/662386
+	"""
+	def __init__(self, repo, spawn_kwargs):
+		"""
+		@param repo: repository configuration
+		@type repo: portage.repository.config.RepoConfig
+		@param spawn_kwargs: keyword arguments supported by the
+			portage.process.spawn function
+		@type spawn_kwargs: dict
+		"""
+		raise NotImplementedError
+
+	@coroutine
+	def init_update(self):
+		"""
+		Create an update directory as a destination to sync updates to.
+		The directory will be populated with files from the previous
+		immutable snapshot, if available. Note that this directory
+		may contain hardlinks that reference files in the previous
+		immutable snapshot, so these files should not be modified
+		(tools like rsync and git normally break hardlinks when
+		files need to be modified).
+
+		@rtype: str
+		@return: path of directory to update, populated with files from
+			the previous snapshot if available
+		"""
+		raise NotImplementedError
+
+	@property
+	def current_update(self):
+		"""
+		Get the current update directory which would have been returned
+		from the most recent call to the init_update method. This raises
+		RepoStorageException if the init_update method has not been
+		called.
+
+		@rtype: str
+		@return: path of directory to update
+		"""
+		raise NotImplementedError
+
+	@coroutine
+	def commit_update(self):
+		"""
+		Commit the current update directory, so that is becomes the
+		latest immutable snapshot.
+		"""
+		raise NotImplementedError
+
+	@coroutine
+	def abort_update(self):
+		"""
+		Delete the current update directory. If there was not an update
+		in progress, or it has already been committed, then this has
+		no effect.
+		"""
+		raise NotImplementedError
+
+	@coroutine
+	def garbage_collection(self):
+		"""
+		Remove expired snapshots.
+		"""
+		raise NotImplementedError
diff --git a/lib/portage/sync/controller.py b/lib/portage/sync/controller.py
index 3bccf6f74..bf5750f7f 100644
--- a/lib/portage/sync/controller.py
+++ b/lib/portage/sync/controller.py
@@ -327,6 +327,7 @@ class SyncManager(object):
 		# override the defaults when sync_umask is set
 		if repo.sync_umask is not None:
 			spawn_kwargs["umask"] = int(repo.sync_umask, 8)
+		spawn_kwargs.setdefault("umask", 0o022)
 		self.spawn_kwargs = spawn_kwargs
 
 		if self.usersync_uid is not None:
diff --git a/lib/portage/sync/modules/rsync/rsync.py b/lib/portage/sync/modules/rsync/rsync.py
index 56e38631e..17b1b9e7b 100644
--- a/lib/portage/sync/modules/rsync/rsync.py
+++ b/lib/portage/sync/modules/rsync/rsync.py
@@ -59,55 +59,6 @@ class RsyncSync(NewBase):
 	def __init__(self):
 		NewBase.__init__(self, "rsync", RSYNC_PACKAGE_ATOM)
 
-	def _select_download_dir(self):
-		'''
-		Select and return the download directory. It's desirable to be able
-		to create shared hardlinks between the download directory to the
-		normal repository, and this is facilitated by making the download
-		directory be a subdirectory of the normal repository location
-		(ensuring that no mountpoints are crossed). Shared hardlinks are
-		created by using the rsync --link-dest option.
-
-		Since the download is initially unverified, it is safest to save
-		it in a quarantine directory. The quarantine directory is also
-		useful for making the repository update more atomic, so that it
-		less likely that normal repository location will be observed in
-		a partially synced state.
-
-		This method returns a quarantine directory if sync-allow-hardlinks
-		is enabled in repos.conf, and otherwise it returne the normal
-		repository location.
-		'''
-		if self.repo.sync_allow_hardlinks:
-			return os.path.join(self.repo.location, '.tmp-unverified-download-quarantine')
-		else:
-			return self.repo.location
-
-	def _commit_download(self, download_dir):
-		'''
-		Commit changes from download_dir if it does not refer to the
-		normal repository location.
-		'''
-		exitcode = 0
-		if self.repo.location != download_dir:
-			rsynccommand = [self.bin_command] + self.rsync_opts + self.extra_rsync_opts
-			rsynccommand.append('--exclude=/%s' % os.path.basename(download_dir))
-			rsynccommand.append('%s/' % download_dir.rstrip('/'))
-			rsynccommand.append('%s/' % self.repo.location)
-			exitcode = portage.process.spawn(rsynccommand, **self.spawn_kwargs)
-
-		return exitcode
-
-	def _remove_download(self, download_dir):
-		"""
-		Remove download_dir if it does not refer to the normal repository
-		location.
-		"""
-		exitcode = 0
-		if self.repo.location != download_dir:
-			exitcode = subprocess.call(['rm', '-rf', download_dir])
-		return exitcode
-
 	def update(self):
 		'''Internal update function which performs the transfer'''
 		opts = self.options.get('emerge_config').opts
@@ -143,8 +94,8 @@ class RsyncSync(NewBase):
 			self.extra_rsync_opts.extend(portage.util.shlex_split(
 				self.repo.module_specific_options['sync-rsync-extra-opts']))
 
-		download_dir = self._select_download_dir()
 		exitcode = 0
+		verify_failure = False
 
 		# Process GLEP74 verification options.
 		# Default verification to 'no'; it's enabled for ::gentoo
@@ -240,10 +191,14 @@ class RsyncSync(NewBase):
 				self.proto = "file"
 				dosyncuri = syncuri[7:]
 				unchanged, is_synced, exitcode, updatecache_flg = self._do_rsync(
-					dosyncuri, timestamp, opts, download_dir)
+					dosyncuri, timestamp, opts)
 				self._process_exitcode(exitcode, dosyncuri, out, 1)
-				if exitcode == 0 and not unchanged:
-					self._commit_download(download_dir)
+				if exitcode == 0:
+					if unchanged:
+						self.repo_storage.abort_update()
+					else:
+						self.repo_storage.commit_update()
+						self.repo_storage.garbage_collection()
 				return (exitcode, updatecache_flg)
 
 			retries=0
@@ -375,7 +330,7 @@ class RsyncSync(NewBase):
 					dosyncuri = dosyncuri[6:].replace('/', ':/', 1)
 
 				unchanged, is_synced, exitcode, updatecache_flg = self._do_rsync(
-					dosyncuri, timestamp, opts, download_dir)
+					dosyncuri, timestamp, opts)
 				if not unchanged:
 					local_state_unchanged = False
 				if is_synced:
@@ -390,6 +345,7 @@ class RsyncSync(NewBase):
 					# exit loop
 					exitcode = EXCEEDED_MAX_RETRIES
 					break
+
 			self._process_exitcode(exitcode, dosyncuri, out, maxretries)
 
 			if local_state_unchanged:
@@ -397,6 +353,8 @@ class RsyncSync(NewBase):
 				# in this case, so refer gemato to the normal repository
 				# location.
 				download_dir = self.repo.location
+			else:
+				download_dir = self.download_dir
 
 			# if synced successfully, verify now
 			if exitcode == 0 and self.verify_metamanifest:
@@ -448,14 +406,18 @@ class RsyncSync(NewBase):
 								% (e,),
 								level=logging.ERROR, noiselevel=-1)
 						exitcode = 1
+						verify_failure = True
 
 			if exitcode == 0 and not local_state_unchanged:
-				exitcode = self._commit_download(download_dir)
+				self.repo_storage.commit_update()
+				self.repo_storage.garbage_collection()
 
 			return (exitcode, updatecache_flg)
 		finally:
-			if exitcode == 0:
-				self._remove_download(download_dir)
+			# Don't delete the update if verification failed, in case
+			# the cause needs to be investigated.
+			if not verify_failure:
+				self.repo_storage.abort_update()
 			if openpgp_env is not None:
 				openpgp_env.close()
 
@@ -594,7 +556,7 @@ class RsyncSync(NewBase):
 		return rsync_opts
 
 
-	def _do_rsync(self, syncuri, timestamp, opts, download_dir):
+	def _do_rsync(self, syncuri, timestamp, opts):
 		updatecache_flg = False
 		is_synced = False
 		if timestamp != 0 and "--quiet" not in opts:
@@ -720,11 +682,6 @@ class RsyncSync(NewBase):
 				# actual sync
 				command = rsynccommand[:]
 
-				if self.repo.location != download_dir:
-					# Use shared hardlinks for files that are identical
-					# in the previous snapshot of the repository.
-					command.append('--link-dest=%s' % self.repo.location)
-
 				submodule_paths = self._get_submodule_paths()
 				if submodule_paths:
 					# The only way to select multiple directories to
@@ -738,7 +695,7 @@ class RsyncSync(NewBase):
 				else:
 					command.append(syncuri + "/")
 
-				command.append(download_dir)
+				command.append(self.download_dir)
 
 				exitcode = None
 				try:
diff --git a/lib/portage/sync/syncbase.py b/lib/portage/sync/syncbase.py
index ce69a4fc0..1d2a00b7c 100644
--- a/lib/portage/sync/syncbase.py
+++ b/lib/portage/sync/syncbase.py
@@ -15,6 +15,7 @@ import portage
 from portage.util import writemsg_level
 from portage.util._eventloop.global_event_loop import global_event_loop
 from portage.util.backoff import RandomExponentialBackoff
+from portage.util.futures._sync_decorator import _sync_methods
 from portage.util.futures.retry import retry
 from portage.util.futures.executor.fork import ForkExecutor
 from . import _SUBMODULE_PATH_MAP
@@ -40,6 +41,8 @@ class SyncBase(object):
 		self.repo = None
 		self.xterm_titles = None
 		self.spawn_kwargs = None
+		self.repo_storage = None
+		self._download_dir = None
 		self.bin_command = None
 		self._bin_command = bin_command
 		self.bin_pkg = bin_pkg
@@ -72,7 +75,35 @@ class SyncBase(object):
 		self.repo = self.options.get('repo', None)
 		self.xterm_titles = self.options.get('xterm_titles', False)
 		self.spawn_kwargs = self.options.get('spawn_kwargs', None)
+		storage_cls = portage.load_mod(self._select_storage_module())
+		self.repo_storage = _sync_methods(storage_cls(self.repo, self.spawn_kwargs))
 
+	def _select_storage_module(self):
+		'''
+		Select an appropriate implementation of RepoStorageInterface, based
+		on repos.conf settings.
+
+		@rtype: str
+		@return: name of the selected repo storage constructor
+		'''
+		if self.repo.sync_allow_hardlinks:
+			mod_name = 'portage.repository.storage.hardlink_quarantine.HardlinkQuarantineRepoStorage'
+		else:
+			mod_name = 'portage.repository.storage.inplace.InplaceRepoStorage'
+		return mod_name
+
+	@property
+	def download_dir(self):
+		"""
+		Get the path of the download directory, where the repository
+		update is staged. The directory is initialized lazily, since
+		the repository might already be at the latest revision, and
+		there may be some cost associated with the directory
+		initialization.
+		"""
+		if self._download_dir is None:
+			self._download_dir = self.repo_storage.init_update()
+		return self._download_dir
 
 	def exists(self, **kwargs):
 		'''Tests whether the repo actually exists'''
-- 
2.16.4