Don't quarantine on read_metadata ENOENT

An operation that removes an existing .ts or .meta out from under another
concurrent operation at the right point can cause the whole object to be
needlessly quarantined.

Closes-Bug: #1451520

Change-Id: I37d660199e54411d0610889f9ee230b13747244b

swift/obj/diskfile.py

@@ -117,13 +117,17 @@ def read_metadata(fd):
             metadata += xattr.getxattr(fd, '%s%s' % (METADATA_KEY,
                                                      (key or '')))
             key += 1
-    except IOError as e:
+    except (IOError, OSError) as e:
         for err in 'ENOTSUP', 'EOPNOTSUPP':
             if hasattr(errno, err) and e.errno == getattr(errno, err):
                 msg = "Filesystem at %s does not support xattr" % \
                       _get_filename(fd)
                 logging.exception(msg)
                 raise DiskFileXattrNotSupported(e)
+        if e.errno == errno.ENOENT:
+            raise DiskFileNotExist()
+        # TODO: we might want to re-raise errors that don't denote a missing
+        # xattr here.  Seems to be ENODATA on linux and ENOATTR on BSD/OSX.
     return pickle.loads(metadata)
 
 
@@ -1590,6 +1594,8 @@ class DiskFile(object):
         # file if we have one
         try:
             return read_metadata(source)
+        except (DiskFileXattrNotSupported, DiskFileNotExist):
+            raise
         except Exception as err:
             raise self._quarantine(
                 quarantine_filename,

test/unit/obj/test_server.py

@@ -4860,6 +4860,51 @@ class TestObjectController(unittest.TestCase):
             self.assertEquals(resp.status_int, 503)
             self.assertFalse(os.path.isdir(object_dir))
 
+    def test_race_doesnt_quarantine(self):
+        existing_timestamp = normalize_timestamp(time())
+        delete_timestamp = normalize_timestamp(time() + 1)
+        put_timestamp = normalize_timestamp(time() + 2)
+
+        # make a .ts
+        req = Request.blank(
+            '/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'DELETE'},
+            headers={'X-Timestamp': existing_timestamp})
+        req.get_response(self.object_controller)
+
+        # force a PUT between the listdir and read_metadata of a DELETE
+        put_once = [False]
+        orig_listdir = os.listdir
+
+        def mock_listdir(path):
+            listing = orig_listdir(path)
+            if not put_once[0]:
+                put_once[0] = True
+                req = Request.blank(
+                    '/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+                    headers={'X-Timestamp': put_timestamp,
+                             'Content-Length': '9',
+                             'Content-Type': 'application/octet-stream'})
+                req.body = 'some data'
+                resp = req.get_response(self.object_controller)
+                self.assertEquals(resp.status_int, 201)
+            return listing
+
+        with mock.patch('os.listdir', mock_listdir):
+            req = Request.blank(
+                '/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'DELETE'},
+                headers={'X-Timestamp': delete_timestamp})
+            resp = req.get_response(self.object_controller)
+            self.assertEquals(resp.status_int, 404)
+
+        qdir = os.path.join(self.testdir, 'sda1', 'quarantined')
+        self.assertFalse(os.path.exists(qdir))
+
+        req = Request.blank('/sda1/p/a/c/o',
+                            environ={'REQUEST_METHOD': 'HEAD'})
+        resp = req.get_response(self.object_controller)
+        self.assertEquals(resp.status_int, 200)
+        self.assertEquals(resp.headers['X-Timestamp'], put_timestamp)
+
 
 @patch_policies(test_policies)
 class TestObjectServer(unittest.TestCase):