PR#794: Work around race in add_external_rpm

Merges #794
https://pagure.io/koji/pull-request/794

Fixes: #788
https://pagure.io/koji/issue/788
rpm from external repo can hit db constraint

hub/kojihub.py

@@ -5730,7 +5730,7 @@ def add_external_rpm(rpminfo, external_repo, strict=True):
 
     # [!] Calling function should perform access checks
 
-    #sanity check rpminfo
+    # sanity check rpminfo
     dtypes = (
         ('name', basestring),
         ('version', basestring),
@@ -5746,37 +5746,49 @@ def add_external_rpm(rpminfo, external_repo, strict=True):
         if not isinstance(rpminfo[field], allowed):
             #this will catch unwanted NULLs
             raise koji.GenericError("Invalid value for %s: %r" % (field, rpminfo[field]))
-    #TODO: more sanity checks for payloadhash
+    # strip extra fields
+    rpminfo = dslice(rpminfo, [x[0] for x in dtypes])
+    # TODO: more sanity checks for payloadhash
 
-    #Check to see if we have it
-    data = rpminfo.copy()
-    data['location'] = external_repo
-    previous = get_rpm(data, strict=False)
+    def check_dup():
+        # Check to see if we have it
+        data = rpminfo.copy()
+        data['location'] = external_repo
+        previous = get_rpm(data, strict=False)
+        if previous:
+            disp = "%(name)s-%(version)s-%(release)s.%(arch)s@%(external_repo_name)s" % previous
+            if strict:
+                raise koji.GenericError("external rpm already exists: %s" % disp)
+            elif data['payloadhash'] != previous['payloadhash']:
+                raise koji.GenericError("hash changed for external rpm: %s (%s -> %s)" \
+                        % (disp, previous['payloadhash'], data['payloadhash']))
+            else:
+                return previous
+
+    previous = check_dup()
     if previous:
-        disp = "%(name)s-%(version)s-%(release)s.%(arch)s@%(external_repo_name)s" % previous
-        if strict:
-            raise koji.GenericError("external rpm already exists: %s" % disp)
-        elif data['payloadhash'] != previous['payloadhash']:
-            raise koji.GenericError("hash changed for external rpm: %s (%s -> %s)" \
-                    % (disp, previous['payloadhash'], data['payloadhash']))
-        else:
+        return previous
+
+    # add rpminfo entry
+    data = rpminfo.copy()
+    data['external_repo_id'] = get_external_repo_id(external_repo, strict=True)
+    data['id'] = nextval('rpminfo_id_seq')
+    data['build_id'] = None
+    data['buildroot_id'] = None
+    insert = InsertProcessor('rpminfo', data=data)
+    savepoint = Savepoint('pre_insert')
+    try:
+        insert.execute()
+    except Exception:
+        # if this failed, it likely duplicates one just inserted
+        # see: https://pagure.io/koji/issue/788
+        savepoint.rollback()
+        previous = check_dup()
+        if previous:
             return previous
+        raise
 
-    #add rpminfo entry
-    rpminfo['external_repo_id'] = get_external_repo_id(external_repo, strict=True)
-    rpminfo['id'] = _singleValue("""SELECT nextval('rpminfo_id_seq')""")
-    q = """INSERT INTO rpminfo (id, build_id, buildroot_id,
-            name, version, release, epoch, arch,
-            external_repo_id,
-            payloadhash, size, buildtime)
-    VALUES (%(id)i, NULL, NULL,
-            %(name)s, %(version)s, %(release)s, %(epoch)s, %(arch)s,
-            %(external_repo_id)i,
-            %(payloadhash)s, %(size)i, %(buildtime)i)
-    """
-    _dml(q, rpminfo)
-
-    return get_rpm(rpminfo['id'])
+    return get_rpm(data['id'])
 
 def import_build_log(fn, buildinfo, subdir=None):
     """Move a logfile related to a build to the right place"""
@@ -7385,6 +7397,16 @@ def nextval(sequence):
     return _singleValue("SELECT nextval(%(sequence)s)", data, strict=True)
 
 
+class Savepoint(object):
+
+    def __init__(self, name):
+        self.name = name
+        _dml("SAVEPOINT %s" % name, {})
+
+    def rollback(self):
+        _dml("ROLLBACK TO SAVEPOINT %s" % self.name, {})
+
+
 def parse_json(value, desc=None, errstr=None):
     if value is None:
         return value

tests/test_hub/test_add_external_rpm.py

@@ -0,0 +1,158 @@
+import mock
+import unittest
+
+import koji
+import kojihub
+
+
+IP = kojihub.InsertProcessor
+
+
+class FakeException(Exception):
+    pass
+
+
+class TestAddExternalRPM(unittest.TestCase):
+
+    def setUp(self):
+        self.get_rpm = mock.patch('kojihub.get_rpm').start()
+        self.get_external_repo_id = mock.patch('kojihub.get_external_repo_id').start()
+        self.nextval = mock.patch('kojihub.nextval').start()
+        self.Savepoint = mock.patch('kojihub.Savepoint').start()
+        self.InsertProcessor = mock.patch('kojihub.InsertProcessor',
+                side_effect=self.getInsert).start()
+        self.inserts = []
+        self.insert_execute = mock.MagicMock()
+
+        self.rpminfo = {
+                'name': 'NAME',
+                'version': 'VERSION',
+                'release': 'RELEASE',
+                'epoch': None,
+                'arch': 'noarch',
+                'payloadhash': 'fakehash',
+                'size': 42,
+                'buildtime': 0,
+                }
+        self.repo = 'myrepo'
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def getInsert(self, *args, **kwargs):
+        insert = IP(*args, **kwargs)
+        insert.execute = self.insert_execute
+        self.inserts.append(insert)
+        return insert
+
+    def test_add_ext_rpm(self):
+        self.get_rpm.return_value = None
+        self.get_external_repo_id.return_value = mock.sentinel.repo_id
+        self.nextval.return_value = mock.sentinel.rpm_id
+
+        # call it
+        kojihub.add_external_rpm(self.rpminfo, self.repo)
+
+        self.assertEqual(len(self.inserts), 1)
+        insert = self.inserts[0]
+        self.assertEqual(insert.data['external_repo_id'], mock.sentinel.repo_id)
+        self.assertEqual(insert.data['id'], mock.sentinel.rpm_id)
+        self.assertEqual(insert.table, 'rpminfo')
+
+    def test_add_ext_rpm_bad_data(self):
+        rpminfo = self.rpminfo.copy()
+        del rpminfo['size']
+
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_external_rpm(rpminfo, self.repo)
+
+        self.get_rpm.assert_not_called()
+        self.nextval.assert_not_called()
+        self.assertEqual(len(self.inserts), 0)
+
+        rpminfo = self.rpminfo.copy()
+        rpminfo['size'] = ['invalid type']
+
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_external_rpm(rpminfo, self.repo)
+
+        self.get_rpm.assert_not_called()
+        self.nextval.assert_not_called()
+        self.assertEqual(len(self.inserts), 0)
+
+    def test_add_ext_rpm_dup(self):
+        prev = self.rpminfo.copy()
+        prev['external_repo_id'] = mock.sentinel.repo_id
+        prev['external_repo_name'] = self.repo
+        self.get_rpm.return_value = prev
+        self.get_external_repo_id.return_value = mock.sentinel.repo_id
+
+        # call it (default is strict)
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_external_rpm(self.rpminfo, self.repo)
+
+        self.assertEqual(len(self.inserts), 0)
+        self.nextval.assert_not_called()
+
+        # call it without strict
+        ret = kojihub.add_external_rpm(self.rpminfo, self.repo, strict=False)
+
+        self.assertEqual(ret, self.get_rpm.return_value)
+        self.assertEqual(len(self.inserts), 0)
+        self.nextval.assert_not_called()
+
+        # different hash
+        prev['payloadhash'] = 'different hash'
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_external_rpm(self.rpminfo, self.repo, strict=False)
+
+        self.assertEqual(len(self.inserts), 0)
+        self.nextval.assert_not_called()
+
+    def test_add_ext_rpm_dup_late(self):
+        prev = self.rpminfo.copy()
+        prev['external_repo_id'] = mock.sentinel.repo_id
+        prev['external_repo_name'] = self.repo
+        self.get_rpm.side_effect = [None, prev]
+        self.get_external_repo_id.return_value = mock.sentinel.repo_id
+        self.insert_execute.side_effect = FakeException('insert failed')
+
+        # call it (default is strict)
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_external_rpm(self.rpminfo, self.repo)
+
+        self.assertEqual(len(self.inserts), 1)
+        self.nextval.assert_called_once()
+
+        # call it without strict
+        self.inserts[:] = []
+        self.nextval.reset_mock()
+        self.get_rpm.side_effect = [None, prev]
+        ret = kojihub.add_external_rpm(self.rpminfo, self.repo, strict=False)
+
+        self.assertEqual(ret, prev)
+        self.assertEqual(len(self.inserts), 1)
+        self.nextval.assert_called_once()
+
+        # different hash
+        self.inserts[:] = []
+        self.nextval.reset_mock()
+        self.get_rpm.side_effect = [None, prev]
+        prev['payloadhash'] = 'different hash'
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_external_rpm(self.rpminfo, self.repo, strict=False)
+
+        self.assertEqual(len(self.inserts), 1)
+        self.nextval.assert_called_once()
+
+        # no dup after failed insert
+        self.inserts[:] = []
+        self.nextval.reset_mock()
+        self.get_rpm.side_effect = [None, None]
+        with self.assertRaises(FakeException):
+            kojihub.add_external_rpm(self.rpminfo, self.repo, strict=False)
+
+        self.assertEqual(len(self.inserts), 1)
+        self.nextval.assert_called_once()
+
+

tests/test_hub/test_savepoint.py

@@ -0,0 +1,22 @@
+import mock
+import unittest
+
+import kojihub
+
+
+class TestSavepoint(unittest.TestCase):
+
+    def setUp(self):
+        self.dml = mock.patch('kojihub._dml').start()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def test_savepoint(self):
+        sp = kojihub.Savepoint('some_name')
+        self.assertEqual(sp.name, 'some_name')
+        self.dml.assert_called_once_with('SAVEPOINT some_name', {})
+
+        self.dml.reset_mock()
+        sp.rollback()
+        self.dml.assert_called_once_with('ROLLBACK TO SAVEPOINT some_name', {})